Forem: Michiel

Stop Writing Boilerplate for File Watching in Python

Michiel — Mon, 12 Jan 2026 14:27:48 +0000

File shows up → do something. Why is this still hard?

I've built more file-processing pipelines than I can count. Media ingest systems, config reloaders, data pipelines, upload processors. They all start the same way: watch a folder, react to changes.

And every single time, I end up writing the same boilerplate:

from watchfiles import watch

for changes in watch('./inbox'):
    for change_type, path in changes:
        if path.endswith('.json'):
            if change_type == Change.added:
                process_new_file(path)
            elif change_type == Change.modified:
                reload_config(path)
            # ... and so on

It works. But it's tedious. The logic for what you're watching gets tangled with how you're watching it.

I wanted something simpler.

FlowWatch

FlowWatch is a tiny ergonomic layer on top of watchfiles that makes it easy to build file-driven workflows using simple decorators and a pretty Rich + Typer powered CLI.

Instead of wiring watchfiles.watch() manually in every project, you declare:

what folder(s) you want to watch
which patterns you care about (e.g. *.mxf, *.json)
which function should run for a given event (created / modified / deleted)

FlowWatch takes care of:

subscribing to all roots in a single watcher loop
debouncing and recursive watching
dispatching events to handlers with a small thread pool
optional processing of existing files on startup
nicely formatted logs and a CLI overview of registered handlers
real-time web dashboard for monitoring events

Installation

FlowWatch is published as a normal Python package.

# Using uv (recommended)
uv add flowwatch

# Or with pip
pip install flowwatch

Optional Extras

# Standalone dashboard (Starlette + uvicorn)
uv

…

View on GitHub

What If File Watching Looked Like This?

from flowwatch import on_created, on_modified, run

@on_created("./inbox", pattern="*.json")
def handle_new_json(event):
    print(f"New JSON file: {event.path}")
    process_upload(event.path)

@on_modified("./config", pattern="*.yaml")
def reload_config(event):
    print(f"Config changed: {event.path}")
    app.reload_settings()

@on_created("./media", pattern="*.mxf", process_existing=True)
def ingest_media(event):
    print(f"New media file: {event.path}")
    start_transcode(event.path)

run()

That's FlowWatch — a decorator-first layer on top of watchfiles that makes file-driven workflows readable and maintainable.

Your intent is obvious from the code. No wiring. No boilerplate.

Why I Built This

I work in broadcast technology where file-based workflows are everywhere. Media files land in watch folders. Metadata sidecars appear alongside them. Config files change.

Every project needed the same pattern:

Watch one or more directories
Filter by file extension
Route to different handlers based on event type
Maybe process files that already exist on startup

I kept copying the same wrapper code between projects. FlowWatch is that code — extracted into a proper library.

Core Concepts

The FileEvent Object

Every handler receives a FileEvent with everything you need:

@on_created("./uploads", pattern="*.pdf")
def handle_pdf(event):
    print(event.path)        # pathlib.Path to the file
    print(event.root)        # the folder being watched
    print(event.pattern)     # the pattern that matched
    print(event.change)      # watchfiles.Change enum

    # Convenience properties
    if event.is_created:
        process_new_file(event.path)

Four Decorators, Four Intents

@on_created(root, pattern="*.txt")    # New files
@on_modified(root, pattern="*.json")  # Changed files
@on_deleted(root, pattern="*.tmp")    # Removed files
@on_any(root, pattern="*.*")          # All events

Process Existing Files on Startup

Set process_existing=True and FlowWatch will scan the directory on startup:

@on_created("./queue", pattern="*.job", process_existing=True)
def process_job(event):
    # Runs for files already in ./queue when you start
    # Then continues watching for new ones
    execute_job(event.path)

No more "catch up on what we missed" logic.

A Real-Time Dashboard

FlowWatch ships with an optional web dashboard:

Features:

Live event counters — created, modified, deleted at a glance
Watched directories — see exactly what folders are being monitored
Recent activity feed — every file event with status, path, and timestamp

Standalone Dashboard

pip install flowwatch[dashboard]

from flowwatch import on_created, run_with_dashboard

@on_created("./inbox", pattern="*.json")
def handle_json(event):
    process_file(event.path)

run_with_dashboard(host="0.0.0.0", port=8000)

FastAPI Integration

pip install flowwatch[fastapi]

from fastapi import FastAPI
from flowwatch import on_created
from flowwatch.fastapi import create_flowwatch_router

app = FastAPI()

@on_created("./uploads", pattern="*.*")
def handle_upload(event):
    process_upload(event.path)

app.include_router(create_flowwatch_router(), prefix="/flowwatch")

The CLI

FlowWatch ships with a Typer + Rich CLI:

flowwatch run myproject.watchers

This imports your module, discovers all handlers, and shows you a formatted table:

┏━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━┓
┃ Handler          ┃ Root             ┃ Events  ┃ Pattern  ┃
┡━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━┩
│ handle_new_json  │ ./inbox          │ created │ *.json   │
│ reload_config    │ ./config         │ modified│ *.yaml   │
│ ingest_media     │ ./media          │ created │ *.mxf    │
└──────────────────┴──────────────────┴─────────┴──────────┘

Options:

flowwatch run myproject.watchers \
  --debounce 500 \
  --max-workers 8 \
  --no-recursive \
  --log-level DEBUG

Real-World Example: Media Ingest

Here's a pattern I use in production:

from pathlib import Path
from flowwatch import FileEvent, on_created, on_deleted, run

INBOX = Path("/media/incoming")

@on_created(str(INBOX), pattern="*.mxf", process_existing=True)
def ingest_media(event: FileEvent) -> None:
    """New media file arrived — start processing."""
    print(f"[ingest] New media: {event.path.name}")

    # Check for sidecar metadata
    sidecar = event.path.with_suffix(".json")
    if sidecar.exists():
        start_full_ingest(event.path, sidecar)
    else:
        queue_pending(event.path)

@on_created(str(INBOX), pattern="*.json")
def handle_sidecar(event: FileEvent) -> None:
    """Metadata sidecar arrived — check if media is waiting."""
    media_file = event.path.with_suffix(".mxf")
    if is_pending(media_file):
        start_full_ingest(media_file, event.path)

if __name__ == "__main__":
    run()

Docker-Ready

FlowWatch runs great as a worker container:

services:
  backend:
    build: ./backend
    volumes:
      - media:/media

  flowwatch:
    build: ./backend
    command: flowwatch run myproject.watchers
    volumes:
      - media:/media
    restart: unless-stopped

volumes:
  media:

When to Use FlowWatch

Perfect for:

File ingest pipelines
Config file reloaders
Upload processing queues
Build systems and asset pipelines
Any "file arrives → do something" workflow

Use something else if:

You need distributed file watching across machines
You need complex DAG-based orchestration (use Airflow, Prefect)

FlowWatch is intentionally simple — a thin layer over watchfiles, not a workflow engine.

Installation

# Core library
pip install flowwatch

# With standalone dashboard (Starlette + uvicorn)
pip install flowwatch[dashboard]

# With FastAPI integration
pip install flowwatch[fastapi]

# All features
pip install flowwatch[all]

Or with uv:

uv add flowwatch
uv add flowwatch --extra dashboard
uv add flowwatch --extra fastapi
uv add flowwatch --extra all

FlowWatch

FlowWatch is a tiny ergonomic layer on top of watchfiles that makes it easy to build file-driven workflows using simple decorators and a pretty Rich + Typer powered CLI.

Instead of wiring watchfiles.watch() manually in every project, you declare:

what folder(s) you want to watch
which patterns you care about (e.g. *.mxf, *.json)
which function should run for a given event (created / modified / deleted)

FlowWatch takes care of:

subscribing to all roots in a single watcher loop
debouncing and recursive watching
dispatching events to handlers with a small thread pool
optional processing of existing files on startup
nicely formatted logs and a CLI overview of registered handlers
real-time web dashboard for monitoring events

Installation

FlowWatch is published as a normal Python package.

# Using uv (recommended)
uv add flowwatch

# Or with pip
pip install flowwatch

Optional Extras

# Standalone dashboard (Starlette + uvicorn)
uv

…

View on GitHub

If it saves you time, give it a ⭐ on GitHub. If you have ideas, open an issue — I'm actively improving it.

What file-watching patterns do you use? Let me know in the comments!

Stop Overengineering Your Python Schedulers — Here's the Cleaner Way

Michiel — Sat, 10 Jan 2026 11:08:22 +0000

You don't need Celery for a cron job.

I've been building Python applications for years, and one pattern I see repeatedly is developers reaching for heavy-duty solutions when they just need to run a function every 5 minutes.

You want to send a daily report? Here comes Redis, Celery, a message broker, worker processes, and suddenly your simple scheduled task requires its own infrastructure documentation.

There's a better way.

The Problem with Python Scheduling Today

Let's be honest about the current landscape:

Celery is fantastic for distributed task queues at scale. But if you're not at scale? You're maintaining a Redis server for a morning email reminder.

APScheduler is capable, but the API feels like it was designed by someone who gets paid by the line of code. Configuring a simple daily job requires understanding triggers, executors, job stores, and more concepts than the actual business logic you're implementing.

Schedule is lightweight and nice, but lacks async support, persistence, and any visibility into what's actually running.

What I wanted was something that felt as natural as writing a decorator, with just enough features to be useful in production — without the overhead.

So I built FastScheduler.

MichielMe / fastscheduler

Decorator-first Python scheduler — cron/interval/at jobs with simple persistence and built-in run history.

FastScheduler

Simple, lightweight task scheduler for Python with async support, timezone handling, cron expressions, and a beautiful real-time dashboard.

If this saves you time, ⭐️ the repo and open an issue for ideas — I'm actively improving it.

Features

🎯 Simple decorator-based API - Schedule tasks in one line
⚡ Async/await support - Native async function support
🕐 Timezone support - Schedule jobs in any timezone
📅 Cron expressions - Complex schedules with cron syntax
💾 Persistent state - Survives restarts, handles missed jobs
🗄️ Database support - SQLite, PostgreSQL, MySQL via SQLModel
🎨 FastAPI dashboard - Beautiful real-time monitoring UI
🔄 Automatic retries - Configurable retry with exponential backoff
⏱️ Job timeouts - Kill long-running jobs automatically
⏸️ Pause/Resume - Control jobs without removing them
📋 Dead Letter Queue - Track and debug failed jobs

Installation

# Basic installation
pip install fastscheduler
# With FastAPI dashboard
pip install fastscheduler[fastapi]

…

View on GitHub

What If Scheduling Looked Like This?

from fastscheduler import FastScheduler

scheduler = FastScheduler()

@scheduler.every(10).seconds
def check_inventory():
    print("Checking stock levels...")

@scheduler.daily.at("09:00")
async def send_morning_report():
    await generate_and_send_report()

@scheduler.cron("0 9 * * MON-FRI")
def weekday_standup_reminder():
    notify_team("Standup in 15 minutes!")

scheduler.start()

That's it. No configuration files. No separate worker processes. No message broker.

Just Python.

Key Features

🔄 Persistence That Actually Works

FastScheduler keeps state on disk. When your app restarts, it knows which jobs exist, when they last ran, and what they missed.

scheduler = FastScheduler(
    state_file="scheduler.json",
    history_retention_days=7,
    max_history=5000
)

🎨 A Dashboard You'll Actually Use

If you're using FastAPI, adding a real-time dashboard takes three lines:

from fastapi import FastAPI
from fastscheduler import FastScheduler
from fastscheduler.fastapi_integration import create_scheduler_routes

app = FastAPI()
scheduler = FastScheduler()

app.include_router(create_scheduler_routes(scheduler))

Now visit /scheduler/ and you get:

Real-time job status with countdown timers
Execution history with success/failure indicators
One-click pause, resume, and cancel
Dead letter queue for debugging failures
Live updates via SSE — no page refresh needed

🌍 Timezones Without the Headaches

@scheduler.daily.at("09:00", tz="America/New_York")
def east_coast_morning():
    send_market_open_alert()

@scheduler.cron("0 9 * * MON-FRI").tz("Asia/Tokyo")
def tokyo_market_open():
    fetch_nikkei_data()

💪 Production-Ready Failure Handling

Timeouts kill jobs that run too long:

@scheduler.every(5).minutes.timeout(30)
def quick_health_check():
    check_all_services()

Retries with exponential backoff:

@scheduler.hourly.at(":15").retries(3)
def sync_with_external_api():
    fetch_and_store_data()

Dead Letter Queue captures failures:

failed_jobs = scheduler.get_dead_letters(limit=100)
for failure in failed_jobs:
    print(f"{failure['func_name']} failed: {failure['error']}")

The Full API in 60 Seconds

Interval-based:

@scheduler.every(10).seconds
@scheduler.every(5).minutes
@scheduler.every(2).hours

Time-based:

@scheduler.daily.at("09:00")
@scheduler.hourly.at(":30")
@scheduler.weekly.monday.at("10:00")

Cron expressions:

@scheduler.cron("*/15 * * * *")      # Every 15 minutes
@scheduler.cron("0 9 * * MON-FRI")   # Weekdays at 9 AM

One-time jobs:

@scheduler.once(60)                   # 60 seconds from now
@scheduler.at("2025-12-25 00:00:00")  # Specific datetime

Real-World Example

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastscheduler import FastScheduler
from fastscheduler.fastapi_integration import create_scheduler_routes

scheduler = FastScheduler(state_file="app_scheduler.json", quiet=True)

@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler.start()
    yield
    scheduler.stop(wait=True)

app = FastAPI(lifespan=lifespan)
app.include_router(create_scheduler_routes(scheduler))

@scheduler.every(1).minutes.timeout(10)
def health_check():
    check_all_services()

@scheduler.daily.at("08:00", tz="America/New_York").retries(2)
async def send_daily_digest():
    users = await get_subscribed_users()
    for user in users:
        await send_digest_email(user)

@scheduler.cron("*/15 9-17 * * MON-FRI").tz("America/New_York")
def sync_crm_data():
    fetch_and_merge_crm_updates()

That's a complete background job system. No Redis. No Celery. No separate processes.

When to Use FastScheduler

Perfect for:

Single-application background tasks
Scheduled jobs in web applications
Periodic data synchronization
Automated reports and notifications
Health checks and monitoring

Use something else if:

You need distributed task execution across multiple workers
You're processing millions of jobs per day
You need exactly-once delivery guarantees

Get Started

# Basic
pip install fastscheduler

# With dashboard
pip install fastscheduler[fastapi]

# Everything
pip install fastscheduler[all]

MichielMe / fastscheduler

Decorator-first Python scheduler — cron/interval/at jobs with simple persistence and built-in run history.

FastScheduler

Simple, lightweight task scheduler for Python with async support, timezone handling, cron expressions, and a beautiful real-time dashboard.

If this saves you time, ⭐️ the repo and open an issue for ideas — I'm actively improving it.

Features

🎯 Simple decorator-based API - Schedule tasks in one line
⚡ Async/await support - Native async function support
🕐 Timezone support - Schedule jobs in any timezone
📅 Cron expressions - Complex schedules with cron syntax
💾 Persistent state - Survives restarts, handles missed jobs
🗄️ Database support - SQLite, PostgreSQL, MySQL via SQLModel
🎨 FastAPI dashboard - Beautiful real-time monitoring UI
🔄 Automatic retries - Configurable retry with exponential backoff
⏱️ Job timeouts - Kill long-running jobs automatically
⏸️ Pause/Resume - Control jobs without removing them
📋 Dead Letter Queue - Track and debug failed jobs

Installation

# Basic installation
pip install fastscheduler
# With FastAPI dashboard
pip install fastscheduler[fastapi]

…

View on GitHub

If it saves you time, give it a ⭐ on GitHub. If you have ideas, open an issue — I'm actively improving it.

What scheduling solutions are you currently using? Let me know in the comments!

Forem: Michiel

Stop Writing Boilerplate for File Watching in Python

MichielMe / flowwatch

FlowWatch is a tiny ergonomic layer on top of Watchfiles that makes it easy to build file-driven workflows using simple decorators and a pretty Rich + Typer powered CLI.

FlowWatch

Installation

Optional Extras

What If File Watching Looked Like This?

Why I Built This

Core Concepts

The FileEvent Object

Four Decorators, Four Intents

Process Existing Files on Startup

A Real-Time Dashboard

Standalone Dashboard

FastAPI Integration

The CLI

Real-World Example: Media Ingest

Docker-Ready

When to Use FlowWatch

Installation

MichielMe / flowwatch

FlowWatch is a tiny ergonomic layer on top of Watchfiles that makes it easy to build file-driven workflows using simple decorators and a pretty Rich + Typer powered CLI.

FlowWatch

Installation

Optional Extras

Stop Overengineering Your Python Schedulers — Here's the Cleaner Way

The Problem with Python Scheduling Today

MichielMe / fastscheduler

Decorator-first Python scheduler — cron/interval/at jobs with simple persistence and built-in run history.

FastScheduler

Features

Installation

What If Scheduling Looked Like This?

Key Features

🔄 Persistence That Actually Works

🎨 A Dashboard You'll Actually Use

🌍 Timezones Without the Headaches

💪 Production-Ready Failure Handling

The Full API in 60 Seconds

Real-World Example

When to Use FastScheduler

Get Started

MichielMe / fastscheduler

Decorator-first Python scheduler — cron/interval/at jobs with simple persistence and built-in run history.

FastScheduler

Features

Installation