定义路由
前言
如果接触过 Flask 框架,对 FastAPI 定义路由的方式一定很熟悉,基本一样:
from fastapi import FastAPI
app = FastAPI()
@app.get("/hello")
def hello():
return {"msg": "hello world"}
声明 HTTP 请求方法
FastAPI 支持 @app.get、@app.post 等快捷装饰器,也支持类似 Flask @app.route 的 @app.api_route,一般用来给同一个视图函数配置多个 HTTP 方法:
@app.api_route("/index", methods=["GET", "POST"])
def index():
return {"msg": "hello world"}
同步和异步路由
FastAPI 兼容同步和异步:视图函数是同步函数即为同步路由,是异步函数(async def)即为异步路由。
- 同步路由的并发基于多线程(由线程池执行);
- 异步路由的并发基于同一线程的事件循环。
注意
不建议在异步路由中调用阻塞的同步函数(如 time.sleep、阻塞 IO),否则会阻塞事件循环。
import asyncio
import time
@app.get("/sync")
def sync_route():
time.sleep(2)
return {"msg": "sync"}
@app.get("/async")
async def async_route():
await asyncio.sleep(2)
return {"msg": "async"}
多个装饰器装饰一个视图函数
一个视图函数可以同时支持多个请求地址:
@app.get("/")
@app.get("/index")
def root():
return {"msg": "hello world"}
使用 APIRouter
APIRouter 用于定义路由组,功能类似 Flask 的蓝图、go-gin 的路由组。大型项目通常按业务模块划分路由组,便于 API 的开发和管理。
from fastapi import APIRouter, FastAPI
app = FastAPI()
route_user = APIRouter(prefix="/user", tags=["用户模块"])
route_article = APIRouter(prefix="/article", tags=["文章模块"])
@route_user.get("/")
def user_index():
return "user index"
@route_article.get("/")
def article_index():
return "article index"
app.include_router(route_user)
app.include_router(route_article)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
按模块拆分路由文件
myapp/route/route1.py:
from fastapi import APIRouter
route = APIRouter(prefix="/route1", tags=["第一个路由组"])
@route.get("/hello", summary="route1-hello")
async def hello():
return "hello from route1"
myapp/route/route2.py:
from fastapi import APIRouter
route = APIRouter(prefix="/route2", tags=["第二个路由组"])
@route.get("/hello", summary="route2-hello")
async def hello():
return "hello from route2"
myapp/main.py:
from fastapi import FastAPI
from route import route1, route2
app = FastAPI()
app.include_router(route1.route)
app.include_router(route2.route)
多应用挂载
项目比较庞大时,除了 APIRouter,还可以用主应用挂载子应用的方式划分:
from fastapi import FastAPI
app = FastAPI(title="主应用", description="主应用描述", version="0.1.0")
@app.get("/app", summary="index page")
def index():
return {"msg": "hello world"}
app2 = FastAPI(title="应用2", description="应用2描述", version="0.1.1")
@app2.get("/app", summary="index page")
def index():
return {"msg": "hello world from app2"}
# 挂载其它应用,设置子应用的请求 URL 前缀为 /app2
app.mount("/app2", app=app2, name="app2")
上述示例中,app2 的 OpenAPI 访问地址为 http://127.0.0.1:8000/app2/docs。
如果之前已开发好了 Flask 应用,也可以直接挂载到 FastAPI 中。不过建议要么全用 Flask,要么全用 FastAPI:
from fastapi import FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware
from flask import Flask
app = FastAPI(title="主应用", description="主应用描述", version="0.1.0")
@app.get("/app", summary="index page")
def index():
return {"msg": "hello world"}
flask_app = Flask(__name__)
@flask_app.get("/flask")
def index():
return "greeting from flask"
app.mount("/flaskapp", app=WSGIMiddleware(flask_app), name="flask_app")
路由装饰器参数
from typing import Any
def get(
path: str,
*,
response_model: Any = None, # 响应模型,用于校验/过滤响应内容
status_code: int | None = None, # 响应状态码
tags: list[str] | None = None, # API 文档中的分组标签,可多个
dependencies=None, # 路由级依赖项
summary: str | None = None, # API 文档中接口的展示名称,默认取函数名
description: str | None = None, # API 文档中接口的详细描述
response_description: str = "Successful Response",
responses=None, # 不同状态码对应的响应模型/描述
deprecated: bool | None = None, # API 文档中标记为废弃
operation_id: str | None = None,
response_model_include=None, # 响应中只保留这些字段
response_model_exclude=None, # 响应中排除这些字段(如敏感字段)
response_model_by_alias: bool = True,
response_model_exclude_unset: bool = False, # 是否不返回未赋值字段
response_model_exclude_defaults: bool = False, # 是否不返回使用默认值的字段
response_model_exclude_none: bool = False, # 是否不返回值为 None 的字段
include_in_schema: bool = True, # 是否显示在 API 文档中
response_class=None, # 响应类,默认 JSONResponse
name: str | None = None, # 路由内部名称,用于 reverse 反查 URL
callbacks=None,
openapi_extra=None,
generate_unique_id_function=None,
):
...
注意:
summary是 API 文档中展示给用户的名称;name是路由在框架内部的名称,主要用于request.url_for(name)反向解析 URL,二者用途不同。response_model_exclude/response_model_include接收set[str](或 dict),例如response_model_exclude={"password"}。response_model_exclude_unset、exclude_defaults、exclude_none用于精细控制响应字段是否输出。