Middleware

Tri-phase middleware pipeline: execute logic before, around, or after handlers. Supports priority ordering, signature-based dependency injection, and timeout enforcement.

Execution Order

Update arrives

↓

Before Middlewares

(highest priority first, signature-based DI)

↓

Around Middlewares

(wrap the handler, control execution)

↓

Handler

↓

After Middlewares

(highest priority first, post-processing)

MiddlewareContext v0.5.0 — preferred

In v0.5.0, the preferred calling convention uses a single MiddlewareContext object instead of positional parameter name sniffing. This gives full IDE autocomplete:

from kurigram_addons import MiddlewareContext

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

await app.include_middleware(auth_middleware, kind="before", priority=100)

Fields: ctx.update, ctx.client, ctx.helper (PatchHelper). The classic positional form (update, client, patch_helper) still works.

Phase Types

Phase	Signature	Purpose
`before`	`async fn(ctx: MiddlewareContext)`	Pre-processing: logging, auth checks, rate limiting
`around`	`async fn(next_handler, update)`	Wraps handler: timing, retry, error boundaries
`after`	`async fn(ctx: MiddlewareContext)`	Post-processing: analytics, cleanup, notifications

Quick Examples

Before middleware — logging

async def log_updates(ctx: MiddlewareContext):
    print(f"Received: {type(ctx.update).__name__}")

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

Around middleware — timing

import time

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

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

Priority

Higher priority values execute first. When two middlewares have the same priority, they execute in registration order.

# Priority 10 runs before priority 0
await app.include_middleware(auth_check, kind="before", priority=10)
await app.include_middleware(log_updates, kind="before", priority=0)

Signature-Based DI

Middleware functions receive arguments based on their parameter names. The dispatcher inspects your function signature and injects only what you need:

# MiddlewareContext — preferred (v0.5+)
async def modern_mw(ctx: MiddlewareContext):
    print(type(ctx.update).__name__)

# Classic positional — still supported
async def simple_logger(update):
    print(type(update).__name__)

# Classic with all three — still supported
async def full_middleware(update, client, patch_helper):
    data = await patch_helper.get_data()
    # ...