Introduction
PyWatch is my learning project aimed at understanding how file system monitoring works. We will track file and folder creation, modification, movement, and deletion in real time.
Regular Python scripts work on demand: run → get result → program exits. PyWatch works differently — it “listens on the wire” and reacts to events happening in the operating system.
Setting up the environment
We will work on macOS in VS Code.
Create and activate a virtual environment:
python3 -m venv venv
source venv/bin/activate
Install the watchdog library:
pip install watchdog
If after installation the IDE (VS Code) does not recognize the library, simply restart the editor.
Starting the code
import time
from datetime import datetime as dt
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
Here we import the modules:
timeanddatetime— for working with time;Pathfrom thepathlibmodule — for convenient path handling;ObserverandFileSystemEventHandler— the key classes fromwatchdogthat allow us to monitor file system events.
Settings
WATCH_PATH = "."
We set the path to be monitored. A dot (".") means the current folder where the program is running.
Helper functions
def stamp():
return dt.now().strftime("%Y-%m-%d %H:%M:%S")
def norm(p: str) -> str:
return str(Path(p).resolve())
stamp()returns the current time in a convenient format for logs.norm(p)converts a path to an absolute one, ensuring consistent strings.
Why do we need .resolve()?
The .resolve() method normalizes a path for the OS:
- converts relative paths to absolute ones (
"./file.txt"→"/Users/user/project/file.txt"), - removes “.”, “..” and other redundant elements,
- on some systems it also resolves symbolic links.
Thus, if different versions of the same path enter the program, .resolve() will bring them to a single string. This is important to avoid duplicates and logic errors.
Event handler
class MyHandler(FileSystemEventHandler):
def on_created(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [CREATE] {kind} {norm(event.src_path)}")
def on_modified(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [MODIFY] {kind} {norm(event.src_path)}")
def on_deleted(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [DELETE] {kind} {norm(event.src_path)}")
def on_moved(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [MOVE] {kind} {norm(event.src_path)} -> {norm(event.dest_path)}")
Here we create an observer class.
What does this class do?
FileSystemEventHandler is the base class from the watchdog library. We create our own MyHandler class that inherits from it and override the methods:
on_created— triggered when the OS signals that a file/folder has been created,on_modified— on modification,on_deleted— on deletion,on_moved— on move or rename.
Each method receives an event object, which contains:
event.src_path— the path to the object,event.dest_path— the new path (only for moves),event.is_directory— a flag indicating whether it is a file or a folder.
How does it work at the OS level?
When you create or delete a file, the OS kernel logs the event in the file system. Each platform has its own mechanism:
- macOS uses
FSEvents, - Linux —
inotify, - Windows —
ReadDirectoryChangesW.
These system services track changes and report them to the Observer. Watchdog under the hood connects directly to these APIs.
The Observer receives an event from the OS and passes it to the handler. The handler calls the appropriate method (on_created, on_deleted, etc.), and inside we decide what to do — for example, print a log to the console.
This means that the program does not “loop and check” if changes occurred, but instead receives ready-made signals from the OS. That’s what makes it fast and efficient.
A bit more detail: how the OS tracks file system changes
When we write code in Python and use the watchdog library, we are working at a high level — we just call Observer and Handler. But it is important to understand that under the hood everything depends on the operating system’s own mechanisms.
Each OS has its own interface for tracking file system changes:
- Linux — inotify
This is part of the Linux kernel. When aninotifyfile descriptor is created, the kernel starts sending events for changes in the specified directories.
Events arrive in a queue: “file created”, “file deleted”, “file modified”. Each event contains the type of operation and the path.
Watchdog simply listens to this queue and converts it into familiar Python events (on_created,on_deleted, etc.). - macOS — FSEvents
Here the kernel service File System Events is used. macOS keeps a log of changes for directories. A program subscribes to this log and receives a list of events with timestamps and paths.
Unlikeinotify, FSEvents can batch several events together (for example, a group of changes in a folder) and deliver them at once. - Windows — ReadDirectoryChangesW
This is a system call in WinAPI. The program specifies a directory, and the OS begins sending notifications to a buffer whenever changes occur inside it.
Here you can subscribe to several event types at once: create, delete, move, attribute changes.
General principle:
- The OS maintains an internal log of file system changes.
- The
Observersubscribes to this log via the system API. - When a user or program creates/deletes/edits a file, the kernel records the event in the log.
- Watchdog receives this event and calls the appropriate method (
on_created,on_deleted,on_modified,on_moved).
Thus, our code reacts to real signals from the kernel, instead of simply “looping and checking folder state”. This saves resources and allows instant reactions.
┌─────────────┐ ┌─────────────┐ ┌───────────────┐ ┌────────────────┐
│ OS kernel │ ---> │ Observer │ ---> │ Handler │ ---> │ Your code │
│ (inotify / │ │ (watchdog) │ │ (on_created, │ │ (print, logic │
│ FSEvents / │ │ │ │ on_deleted…) │ │ backups, etc.)│
│ ReadDir...) │ │ │ │ │ │ │
└─────────────┘ └─────────────┘ └───────────────┘ └────────────────┘Running the observer
if __name__ == "__main__":
path = WATCH_PATH
handler = MyHandler()
observer = Observer()
observer.schedule(handler, path, recursive=True)
observer.start()
print(f"Watching {Path(path).resolve()} (Ctrl+C to stop)")
try:
while True:
time.sleep(0.25)
except KeyboardInterrupt:
observer.stop()
observer.join()
What happens here:
- Create an instance of
MyHandler. - Create an
Observer— it handles communication with the OS. - “Assign” the handler to the required folder using
schedule. - Start the observer.
- The program runs in an infinite loop until we stop it with Ctrl+C.
- On exit, it properly shuts down (
observer.stop()andobserver.join()).
Running the program
Save the file, for example as pywatch.py, and run:
python3 pywatch.py
Now try creating or deleting a file in the monitored folder. Events will immediately appear in the console.
Final code
import time
from datetime import datetime
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
WATCH_PATH = "."
def stamp():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def norm(p: str) -> str:
return str(Path(p).resolve())
class MyHandler(FileSystemEventHandler):
def on_created(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [CREATE] {kind} {norm(event.src_path)}")
def on_modified(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [MODIFY] {kind} {norm(event.src_path)}")
def on_deleted(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [DELETE] {kind} {norm(event.src_path)}")
def on_moved(self, event):
kind = "DIR " if event.is_directory else "FILE"
print(f"{stamp()} [MOVE] {kind} {norm(event.src_path)} -> {norm(event.dest_path)}")
if __name__ == "__main__":
path = WATCH_PATH
handler = MyHandler()
observer = Observer()
observer.schedule(handler, path, recursive=True)
observer.start()
print(f"Watching {Path(path).resolve()} (Ctrl+C to stop)")
try:
while True:
time.sleep(0.25)
except KeyboardInterrupt:
observer.stop()
observer.join()