跳到主要内容

定义路由

前言

如果接触过 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_unsetexclude_defaultsexclude_none 用于精细控制响应字段是否输出。

参考