Writing Middleware

Middleware intercepts every update before (before), after (after), or around (around) the handler. Two calling conventions are supported: the classic positional form and the v0.5.0 MiddlewareContext typed form.

MiddlewareContext v0.5.0 — preferred

Annotate the first parameter as MiddlewareContext to receive a single structured object instead of relying on positional name sniffing. Fields: ctx.update, ctx.client, ctx.helper.

from kurigram_addons import MiddlewareContext

async def logging_middleware(ctx: MiddlewareContext):
    user_id = getattr(ctx.update.from_user, "id", "?")
    state   = await ctx.helper.state
    print(f"[{user_id}] state={state}")

async def auth_middleware(ctx: MiddlewareContext):
    user = await db.get_user(ctx.update.from_user.id)
    if not user:
        await ctx.update.reply("Please register first.")
        return   # short-circuit — skip handler

Classic Positional Form

Still supported — the dispatcher sniffs parameter names and injectsupdate, client, and patch_helper by name.

async def logging_middleware(update, client, patch_helper):
    user_id = getattr(update.from_user, "id", "?")
    print(f"[{user_id}] update received")

Before Middleware

async def logging_middleware(ctx: MiddlewareContext):
    print(f"Before handler: {type(ctx.update).__name__}")

await app.include_middleware(logging_middleware, kind="before")

Around Middleware

Wrap the handler call — useful for timing, transactions, or error catching. Must accept and call next_handler.

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

await app.include_middleware(timing_middleware, kind="around")

After Middleware

async def analytics_middleware(ctx: MiddlewareContext):
    await track_event("update_handled", user=ctx.update.from_user.id)

await app.include_middleware(analytics_middleware, kind="after")

Priority

Higher priority runs first. Default is 0.

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

Class-based Middleware

class BanCheckMiddleware:
    def __init__(self, banned_users: set):
        self.banned = banned_users

    async def __call__(self, ctx: MiddlewareContext):
        uid = getattr(ctx.update.from_user, "id", None)
        if uid in self.banned:
            return

ban_check = BanCheckMiddleware(banned_users={99999})
await app.include_middleware(ban_check, kind="before", priority=200)