Writing Middleware

Create custom middleware functions with signature-based dependency injection.

Before Middleware

async def logging_middleware(update, client, patch_helper):
    """Log every incoming update before handlers run."""
    print(f"Update from {update.from_user.id}")

await manager.add_middleware(logging_middleware, kind="before")

Around Middleware

async def timing_middleware(next_handler, update):
    """Measure handler execution time."""
    import time
    start = time.monotonic()
    result = await next_handler(update)
    elapsed = time.monotonic() - start
    print(f"Handler took {elapsed:.3f}s")
    return result

await manager.add_middleware(timing_middleware, kind="around")

After Middleware

async def analytics_middleware(update, client, patch_helper):
    """Send analytics after handler completes."""
    await send_analytics(event="update_handled", user=update.from_user.id)

await manager.add_middleware(analytics_middleware, kind="after")

Priority

Higher priority middleware runs first. Default is 0. Use negative values to run last:

await manager.add_middleware(auth_check, kind="before", priority=100)   # runs first
await manager.add_middleware(logging_mw, kind="before", priority=0)     # runs second
await manager.add_middleware(cleanup_mw, kind="after", priority=-10)    # runs last