PyWatch — First Program

                
                PyWatch — First Program
In this article, we got acquainted with PyWatch — a simple tool for monitoring file system changes using Python. We created a virtual environment on macOS, installed the watchdog library, and wrote our first observer program.

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:

  • time and datetime — for working with time;
  • Path from the pathlib module — for convenient path handling;
  • Observer and FileSystemEventHandler — the key classes from watchdog that 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,
  • Linuxinotify,
  • WindowsReadDirectoryChangesW.

 

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 an inotify file 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.
    Unlike inotify, 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:

  1. The OS maintains an internal log of file system changes.
  2. The Observer subscribes to this log via the system API.
  3. When a user or program creates/deletes/edits a file, the kernel records the event in the log.
  4. 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:

  1. Create an instance of MyHandler.
  2. Create an Observer — it handles communication with the OS.
  3. “Assign” the handler to the required folder using schedule.
  4. Start the observer.
  5. The program runs in an infinite loop until we stop it with Ctrl+C.
  6. On exit, it properly shuts down (observer.stop() and observer.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()