五个可复用的中间件模式
Robyn 的 before_request / after_request 看着简单，但组合起来能覆盖绝大多数横切需求。这篇给出五个能直接搬进项目的写法，以及每个坑在哪。
开始前先记住两条语义（完整说明见 中间件）：
before_request接收并返回 request，可以修改它- 一旦某个
before_request返回了Response，后续中间件和路由处理器都会被跳过，但after_request仍会执行
第二点是实现鉴权短路的基础。
一、请求 ID：让日志串起来
import uuid
from robyn import Request, Response
@app.before_request("/*")
async def attach_request_id(request: Request):
request.headers.set("x-request-id", request.headers.get("x-request-id") or uuid.uuid4().hex)
return request
@app.after_request("/*")
def echo_request_id(request: Request, response: Response):
response.headers.set("x-request-id", request.headers.get("x-request-id"))
return response注意 after_request 可以同时接收 request 和 response——这是把请求侧信息写回响应头的唯一途径。上游如果已经有 x-request-id（比如 CDN 注入的）就直接透传，不要覆盖，否则链路会断。
二、鉴权：用返回 Response 来短路
def parse_token(request):
auth = request.headers.get("authorization")
return auth.removeprefix("Bearer ").strip() if auth else None
@app.before_request("/api/*")
async def require_auth(request: Request):
if request.url.path.startswith("/api/public"):
return request # 放行白名单
token = parse_token(request)
if not token or not verify(token):
return Response(status_code=401, body='{"error": "unauthorized"}',
headers={"Content-Type": "application/json"})
request.headers.set("x-user-id", get_user_id(token))
return request把解析出的用户 ID 写进 request headers 传给下一层，比在处理器里再解一次 token 干净得多。
如果你的鉴权来自 Robyn 自带的身份认证体系，可以直接用 auth_required=True 配合 app.authentication_handlers（见 身份验证），它本质上也是一个 before_request 钩子，可以与自定义中间件叠加使用。
三、限流：别忘了多进程
单进程版本用内存即可：
import time
from collections import defaultdict, deque
WINDOW, LIMIT = 60, 100
_hits: dict[str, deque] = defaultdict(deque)
@app.before_request("/api/*")
async def rate_limit(request: Request):
now = time.time()
q = _hits[request.ip_addr]
while q and now - q[0] > WINDOW:
q.popleft()
if len(q) >= LIMIT:
return Response(status_code=429, body="too many requests",
headers={"Retry-After": str(int(WINDOW))})
q.append(now)
return request但这个版本在 --processes > 1 下会失效：每个进程各自计数，实际放行量是 LIMIT × 进程数，而且同一个 IP 落到不同进程会得到忽紧忽松的体验。生产请换成 Redis：
@app.before_request("/api/*")
async def rate_limit(request: Request):
key = f"rl:{request.ip_addr}:{int(time.time()) // WINDOW}"
count = await redis.incr(key)
if count == 1:
await redis.expire(key, WINDOW)
if count > LIMIT:
return Response(status_code=429, body="too many requests")
return request把 request.url.path 一起拼进 key，就能把窗口从"每 IP"细化到"每 IP 每接口"，避免某个重接口吃光配额。
四、响应统一包装与耗时统计
import time
@app.before_request("/*")
async def start_timer(request: Request):
request.headers.set("x-start-time", str(time.perf_counter()))
return request
@app.after_request("/*")
def add_timing(request: Request, response: Response):
start = float(request.headers.get("x-start-time") or 0)
response.headers.set("x-response-time-ms", f"{(time.perf_counter() - start) * 1000:.2f}")
response.headers.set("x-powered-by", "robyn")
return response高频场景下别在中间件里 print()，同步 IO 会拖慢整条链路；把日志丢进队列异步消费，或直接交给 stdout 采集。
五、全局异常兜底
from robyn import Response
@app.exception
def handle_exception(error):
return Response(status_code=500, body='{"error": "internal server error"}',
headers={"Content-Type": "application/json"})生产环境不要回显 error 的原文，否则异常堆栈会直接暴露给客户端——开发时用 --dev 看就行。
六、组合顺序与性能建议
before_request按注册顺序执行，鉴权类要放在最前，日志类次之中间件是每个请求都跑的，里面做任何同步 IO 都会被放大 N 倍
路径匹配支持
/*通配；不需要全局生效时请写具体的路由前缀，减少无谓调用中间件无法拿到最终 JSON Body 做签名/加密这类需求时，改在单个处理器里显式处理更清晰