从 Flask / FastAPI 迁移到 Robyn
Robyn 的路由写法看起来很像 Flask，注入方式又有点像 FastAPI，很多人因此以为迁移只是改几个装饰器的事。真动起来会发现：真正要改的是读取请求参数的习惯，以及那些基于 ASGI 生态的三方库。下面按迁移顺序拆解。
一、先认清一个前提：Robyn 不是 ASGI 框架
这是迁移前必须接受的事实。Flask/FastAPI（Starlette）都跑在 ASGI 或 WSGI 之上，因而能用大量通用中间件；Robyn 的请求生命周期由 Rust 层处理，Python 只负责业务处理器，所以：
| 原本依赖的东西 | 迁移后 |
|---|---|
| ASGI 中间件（Starlette 生态） | ❌ 不可用，需改用 before_request / after_request |
| Sentry、OpenTelemetry 的 ASGI Instrumentor | ❌ 需用专门的 Robyn 插桩，或自己包一层中间件 |
依赖并关系型的三方扩展（如 fastapi-users） | ❌ 基本无法复用 |
| Pydantic 模型做校验 | ✅ 仍可在处理器里手动调用 |
| Jinja2 模板 | ✅ 官方支持，见 模板 |
| SQLAlchemy / asyncpg 等纯 Python 库 | ✅ 与框架无关，直接迁移 |
判断要不要迁移的经验法则：如果你的项目大量使用了 FastAPI 的 Depends、中间件栈和生态扩展，迁移成本会远高于性能收益；如果项目是薄薄的 REST 层 + 自己的业务逻辑，迁移通常只需要一两天。
二、路由与路径参数
语法对照：
| 场景 | Flask / FastAPI | Robyn |
|---|---|---|
| 声明 | @app.get("/users") | @app.get("/users") |
| 路径参数 | /users/<int:id>、/users/{id} | /users/:id |
| 读取方式 | 函数形参 id | request.params["id"] 或按类型注入的 path_params |
| 蓝图 | Blueprint(prefix="/api") | SubRouter |
Flask 写法 → Robyn 写法：
from robyn import Robyn
app = Robyn(__file__)
@app.get("/users/:id")
async def get_user(request):
user_id = request.params["id"] # 路径参数，永远是字符串
page = request.query_params.get("page") # 查询参数
return {"id": user_id, "page": page}也可以让 Robyn 按类型注解把各部分注入进来，省掉一层 request. 前缀：
from robyn import Request, QueryParams
from robyn.types import PathParams
@app.get("/users/:id")
async def get_user(request: Request, path_params: PathParams, query_params: QueryParams):
return {"id": path_params["id"], "page": query_params.get("page")}注意 Robyn 的路径参数没有类型转换器：:id 抓到的永远是字符串，int(id) 要自己做。
三、请求体与响应
请求体解析统一走 request.json()（非 JSON 会抛 ValueError）：
@app.post("/orders")
async def create_order(request):
data = request.json()
order_id = create_order_in_db(data)
return {"order_id": order_id}返回值会自动转换：字符串 → text/plain，dict/list → JSON。需要自定义状态码时用 Response：
from robyn import Response
@app.post("/orders")
async def create_order(request):
data = request.json()
return Response(
status_code=201,
body='{"status": "created"}',
headers={"Content-Type": "application/json"},
)四、中间件：Flask/FastAPI 的那一套要改写
@app.before_request 接收并返回请求对象；一旦返回 Response，后续中间件和处理器都会被跳过（但 after_request 仍会执行）：
from robyn import Request, Response
@app.before_request("/api/*")
async def require_token(request: Request):
if not request.headers.get("authorization"):
return Response(status_code=401, body="unauthorized")
request.headers.set("x-request-id", gen_id())
return request
@app.after_request("/api/*")
def add_server_header(response: Response):
response.headers.set("x-powered-by", "robyn")
return response全局异常处理替换 FastAPI 的 @app.exception_handler：
from robyn import Response
@app.exception
def handle_exception(error):
return Response(status_code=500, body=f"error msg: {error}")五、依赖注入与校验
FastAPI 的 Depends 在 Robyn 里没有等价物，官方提供了轻量的依赖注入能力（见 依赖注入），更常见的做法是自己<｜hy_place▁holder▁no▁813｜>一层工厂函数：
from functools import wraps
def with_db(func):
@wraps(func)
async def wrapper(request, *args, **kwargs):
kwargs["conn"] = await pool.acquire()
try:
return await func(request, *args, **kwargs)
finally:
await pool.release(kwargs["conn"])
return wrapper
@app.get("/users/:id")
@with_db
async def get_user(request, conn, **kwargs):
return await fetch_user(conn, request.params["id"])入参校验建议用 Pydantic 手动包一层，官方文档里也有 Pydantic 集成 的示例。
六、迁移清单
- 把每个
<var>/{var}改成:var，参数取出方式改成request.params - 逐条替换
before_request/after_request - 移走所有依赖 ASGI 的中间件与工具，逐个找替代
- 确认
sync/async都支持——Robyn 两者都接受，不必一次性全改成协程 - 用
--dev开发，用--processes上生产（详见 生产部署清单） - 别忘了多进程下不共享内存，模块级可变状态需要外置到 Redis 等存储（见 多进程执行）
小结
迁移的收益在 CPU 密集度较低、请求量大的 JSON API 上最明显；而成本几乎全部集中在"替换 ASGI 生态依赖"这一项。如果你只是想要一个更快的 FastAPI，先确认自己用到的扩展有没有 Robyn 版本，比直接动手改代码更重要。