fastapi-路由_fastapi lifespan

FastAPI的APIRouter是一个用于将不同功能模块的端点进行划分的工具，类似于Flask中的蓝图（Blueprint）。通过APIRouter，你可以将应用程序的不同部分组织成独立的路由模块，从而提高代码的可读性和可维护性。

APIRouter允许你在不同的模块中定义路由操作，然后将它们挂载到全局应用程序路由中。这样，你可以将相关的路由操作组织在一起，使代码结构更加清晰，同时便于管理和维护

APIRouter使用

假设你在routers/router_1.py文件下，定义了以下这么一个非常简单的APIRouter：

from fastapi import APIRouter

router = APIRouter()

@router.get("/get")
async def get():
    return {'api': 'get'}
    
@router.post("/post")
async def post():
    return {'api': 'post'}
    
@router.put("/put")
async def put():
    return {'api': 'put'}
    
@router.patch("/patch")
async def patch():
    return {'api': 'patch'}
    
@router.delete("/delete")
async def delete():
    return {'api': 'delete'}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

定义好APIRouter之后，需要在应用程序中将其添加进来，如下：

from fastapi import FastAPI
from routers.router_1 import router

app = FastAPI()
app.include_router(router)
1
2
3
4
5

当我们添加一组路由时，我们需要给这些路由加一个前缀，表示这是同一批路由。我们可以用到prefix参数，如下：

from fastapi import APIRouter

router = APIRouter(prefix="/apis")

@router.get("/get")
async def get():
    return {'api': 'get'}
1
2
3
4
5
6
7

如此，我们可以通过http://localhost:8000/apis/get来请求该路由。

参数介绍

APIRouter常见参数如下：

prefix

prefix为API前缀，接收一个str数据

tags

tags为路由设置标签，接收一个List[str]或者List[Enum]类型

router = APIRouter(tags=["home", "user"])

@router.get("/home")
async def home():
    return {"code": 1}
1
2
3
4
5

在Swagger UI中效果如下：

dependencies

dependencies为依赖项，可用来做一些额外操作(例如权限验证等)，接收一个Sequence[params.Depends]类型。如下：

from fastapi import Header
from fastapi import Depends
from fastapi import APIRouter

async def token_validation(authorization: str = Header(...)):
    print(authorization)

router = APIRouter(dependencies=[Depends(token_validation)])

@router.get("/home")
async def home():
    return {"code": 1}
1
2
3
4
5
6
7
8
9
10
11
12

当我们请求/home这个路由时，如果没有携带Authorization标头，将会抛出422错误，如下：

{
    "detail": [
        {
            "loc": [
                "header",
                "authorization"
            ],
            "msg": "field required",
            "type": "value_error.missing"
        }
    ]
}
1
2
3
4
5
6
7
8
9
10
11
12

default_response_class

default_response_class为默认响应类，接收一个Response类型，默认是JSONResponse。所以当我们视图函数中只需要返回json所支持的类型数据即可，它会包装成JSON响应给到客户端

responses

responses为响应状态示例，接收一个dict类型。如下：

router = APIRouter(responses={404: {"description": "Resource Not Found",
                                    "content": {
                                        "application/json": {
                                            "schema": {
                                                "type": "object",
                                                "properties": {
                                                    "code": {
                                                        "type": "integer",
                                                        "format": "int32"
                                                    }
                                                }
                                            }
                                        }
                                    }}
                              })
@router.get("/home")
async def home():
    return {"code": 1}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

在Swagger UI中效果如下：

上面代码也可以修改成这样：

from fastapi import APIRouter
from pydantic import BaseModel

class NotFoundModel(BaseModel):
    code: int

router = APIRouter(responses={404: {"description": "Resource Not Found",
                                    "model": NotFoundModel}
                              })
@router.get("/home")
async def home():
    return {"code": 1}                 
1
2
3
4
5
6
7
8
9
10
11
12

另外，你也可以修改它默认的200响应示例(但并不推荐这么做)。状态码与对应的字典中支持的字段与格式，可以参考Swagger Editor中responses部分

callbacks

callbacks为回调函数列表，接收List[BaseRoute]类型，在APIRouter中设置似乎不起作用！

routes

routes为路由列表，接收List[routing.BaseRoute]类型

from fastapi import APIRouter
from fastapi.routing import APIRoute

async def home():
    return {"code": 1}

route = APIRoute("/home", endpoint=home, methods=['GET'])
router = APIRouter(routes=[route])
1
2
3
4
5
6
7
8

redirect_slashes

redirect_slashes用于是否检测和重定向url中的斜杠，默认为True。当设置为True时，如果路由为/home，当我们请求/home/时将会返回307状态码，并重定向至/home。同样，如果路由为/home/，而请求/home同样会进行重定向！但是如果设置为False时，将会引发404错误！！值得注意的是，APIRouter中设置不起效果，主要是根据FastAPI初始化的redirect_slashes值来判断

default

default用于处理404未找到错误的默认处理函数，为一个Callable对象，在APIRouter中设置似乎不起作用！

dependency_overrides_provider

dependency_overrides_provider: 仅在FastAPI内部使用，用于处理依赖覆盖。接收任何格式的数据

route_class

route_class为路由类，默认是APIRoute

on_startup

on_startup为启动事件处理程序函数列表，接收一个List[Callable]对象。推荐使用lifespan参数替代

on_shutdown

on_shutdown为关闭事件处理程序函数的列表，接收一个List[Callable]对象。推荐使用lifespan参数替代，使用如下：

async def orm_init():
    print('orm_init')

async def orm_close():
    print('orm_close')

router = APIRouter(on_startup=[orm_init], on_shutdown=[orm_close])
1
2
3
4
5
6
7

程序启动之后将会执行orm_init，而程序关闭之前将会执行orm_close!

lifespan

lifespan为生命周期上下文管理器处理程序。值得注意的是，APIRouter中设置不起效果，主要是根据FastAPI初始化的lifespan值来判断。使用如下：

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def orm_helper(app: APIRouter):
    print('orm init')
    yield
    print('orm close')

app = FastAPI()
1
2
3
4
5
6
7
8
9
10

deprecated

deprecated用于表示是否过期，接收一个bool类型。使用如下：

from fastapi import APIRouter

router = APIRouter(prefix="/apis", tags=['home'], deprecated=True)

@router.get("/get")
async def get():
    return {'api': 'get'}
1
2
3
4
5
6
7

在Swagger UI中效果如下：

include_in_schema

include_in_schema: 是否在Swagger UI中包含该路由，接收一个bool类型，默认为True

generate_unique_id_function

generate_unique_id_function: 生成唯一ID的方法，接收一个Callable对象

方法介绍

route()

route(path, methods, name, include_in_schema)视图方法，使用如下：

from fastapi import APIRouter
from fastapi.responses import JSONResponse

router = APIRouter(tags=['home'])

@router.route("/home", methods=['GET'])
async def home(request):
    return JSONResponse({"code": 1})
1
2
3
4
5
6
7
8

使用该方法的视图，不会在Swagger UI中显示!

add_api_route()

add_api_route(...)用于添加视图，相关参数介绍如下：

• path: 路径，接收一个str类型
• endpoint: 执行函数，接收一个Callable类型
• response_model: 响应模型，接收任何类型数据
• status_code: HTTP状态码，接收一个int类型
• tags: 标签列表，接收一个List[str]或者List[Enum]类型。用法与APIRouter一样
• dependencies: 为依赖项，可用来做一些额外操作(例如权限验证等)。用法与APIRouter一样
• summary: 总结，接收一个str类型
• description: 描述，接收一个str类型
• response_description: 响应说明，接收一个str类型
• responses: 响应状态示例，接收一个dict类型。用法与APIRouter的responses一样
• deprecated: 用于表示是否过期，接收一个bool类型。用法与APIRouter一样
• methods: 请求方法列表，接收一个Set[str]或List[str]类型
• operation_id: 唯一ID，接收一个str类型
• response_model_include: 响应包含哪些字段，接收一个Set或者Dict
• response_model_exclude: 响应不要包含哪些字段，接收一个Set或者Dict
• response_model_by_alias: Response Model是否显示别名，接收一个bool类型True
• response_model_exclude_unset: Response Model是否排除未设置的字段，默认为False
• response_model_exclude_defaults: Response Model是否排除默认的字段，默认为False
• response_model_exclude_none: Response Model是否排除为None的字段，默认为False
• include_in_schema: 是否在Swagger UI中包含该视图，接收一个bool类型，默认为True。用法与APIRouter一样
• response_class: 为默认响应类，接收一个Response类型，默认是JSONResponse。
• name: 为视图名称，接收一个str类型
• route_class_override: 路由类，可以传递一个APIRoute类型，用于覆盖默认的APIRoute
• callbacks: 回调函数列表，不起作用
• openapi_extra: swagger ui中额外的参数
• generate_unique_id_function: 生成唯一ID的方法，接收一个Callable对象。用法与APIRouter一样

add_api_route()使用如下：

from pydantic import Field
from pydantic import BaseModel

class ResponseModel(BaseModel):
    code: int = Field(title='状态码', description='状态码, 1:成功, 0:失败')
    msg: str = Field(title="描述", description='状态码对应描述')

async def create_user():
    resp = ResponseModel(code=1, msg='成功')
    return resp.model_dump()

router.add_api_route('/create', create_user,
                     response_model=ResponseModel,
                     status_code=2000,
                     tags=['user'],
                     summary='创建用户',
                     description='创建用户操作',
                     response_description='创建用户成功',
                     deprecated=True,
                     methods=['GET', 'POST'],
                     response_model_exclude={'msg'})
app.include_router(router)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

在Swagger UI中效果如下：

上图为演示效果，实际操作不建议设置为这样的值！！！

上面API，请求如下:

$ curl --location --request POST 'localhost:8000/create'
{"code":1}      # 因为msg被exclude了
1
2

api_route()

api_route()装饰器方法，其内部实现为调用了add_api_route()方法。所以add_api_route()所支持的参数，api_route()基本上都可以使用。如下：

@router.api_route("/create", response_model=ResponseModel, methods=['GET', 'POST'], summary='创建用户', description='创建用户操作')
async def create_user():
    resp = ResponseModel(code=1, msg='成功')
    return resp.model_dump()
1
2
3
4

get()、post()、put()、delete()、patch()、options()、trace()、head()

装饰器方法，其内部实现为调用了api_route()。所以add_api_route()所支持的参数，它们基本也都可以使用。使用如下：

@router.post("/create", response_model=ResponseModel, summary='创建用户', description='创建用户操作')
async def create_user():
    resp = ResponseModel(code=1, msg='成功')
    return resp.model_dump()
1
2
3
4

该方式也是定义视图最为常见的一种方式！

add_api_websocket_route()

add_api_websocket_route()用于添加websocket视图，相关参数如下:

• path: 路径，接收一个str类型
• endpoint: 执行函数，接收一个Callable类型
• name: 为视图名称，接收一个str类型
• dependencies: 为依赖项，可用来做一些额外操作(例如权限验证等)。用法与APIRouter一样

使用如下：

async def websocket_endpoint(user_id: UUID, websocket: WebSocket):
    print(111, user_id)
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message Text was: {data}")

router.add_api_websocket_route("/{user_id:uuid}", websocket_endpoint)
app.include_router(router)
1
2
3
4
5
6
7
8
9

websocket_route()

websocket_route()为装饰器方法。

websocket()

websocket()为装饰器方法，其内部实现为调用了add_api_websocket_route()方法。

router = APIRouter(tags=['ws'])

@router.websocket("/{user_id:uuid}")
async def websocket_endpoint(user_id: UUID, websocket: WebSocket):
    print(111, user_id)
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message Text was: {data}")

app.include_router(router)
1
2
3
4
5
6
7
8
9
10
11

include_router()

include_router()为添加路由模块。对于多模块应用，这也是最为常见的一种方式。相关参数介绍如下：

• router: 接收一个APIRouter对象
• prefix: 路由前缀
• tags: 标签列表
• dependencies: 依赖项，可用来做一些额外操作(例如权限验证等)，接收一个Sequence[params.Depends]类型
• deprecated: 用于表示是否过期，接收一个bool类型
• include_in_schema: 是否在Swagger UI中包含该视图，接收一个bool类型，默认为True
• default_response_class: 默认响应类，接收一个Response类型，默认是JSONResponse。所以当我们视图函数中只需要返回json所支持的类型数据即可，它会包装成JSON响应给到客户端
• callbacks: 回调函数列表，不起作用
• generate_unique_id_function: 生成唯一ID的方法，接收一个Callable对象

on_event()

on_event()装饰器方法，定义事件监听操作。使用如下：

@router.on_event('startup')
async def startup():
    print('app startup')

app.include_router(router)
1
2
3
4
5

FastAPI

FastAPI可以看做是一个特殊的APIRouter。APIRouter所支持的方法，FastAPI也基本都支持！