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'}
 
定义好APIRouter之后,需要在应用程序中将其添加进来,如下:
from fastapi import FastAPI
from routers.router_1 import router
app = FastAPI()
app.include_router(router)
 
当我们添加一组路由时,我们需要给这些路由加一个前缀,表示这是同一批路由。我们可以用到prefix参数,如下:
from fastapi import APIRouter
router = APIRouter(prefix="/apis")
@router.get("/get")
async def get():
    return {'api': 'get'}
 
如此,我们可以通过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}
 
在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}
 
当我们请求/home这个路由时,如果没有携带Authorization标头,将会抛出422错误,如下:
{
    "detail": [
        {
            "loc": [
                "header",
                "authorization"
            ],
            "msg": "field required",
            "type": "value_error.missing"
        }
    ]
}
 
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}
 
在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}                 
 
另外,你也可以修改它默认的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])
 
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])
 
程序启动之后将会执行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()
 
deprecated
deprecated用于表示是否过期,接收一个bool类型。使用如下:
from fastapi import APIRouter
router = APIRouter(prefix="/apis", tags=['home'], deprecated=True)
@router.get("/get")
async def get():
    return {'api': 'get'}
 
在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})
 
使用该方法的视图,不会在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或者Dictresponse_model_exclude: 响应不要包含哪些字段,接收一个Set或者Dictresponse_model_by_alias: Response Model是否显示别名,接收一个bool类型Trueresponse_model_exclude_unset: Response Model是否排除未设置的字段,默认为Falseresponse_model_exclude_defaults: Response Model是否排除默认的字段,默认为Falseresponse_model_exclude_none: Response Model是否排除为None的字段,默认为Falseinclude_in_schema: 是否在Swagger UI中包含该视图,接收一个bool类型,默认为True。用法与APIRouter一样response_class: 为默认响应类,接收一个Response类型,默认是JSONResponse。name: 为视图名称,接收一个str类型route_class_override: 路由类,可以传递一个APIRoute类型,用于覆盖默认的APIRoutecallbacks: 回调函数列表,不起作用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)
 
在Swagger UI中效果如下:

上图为演示效果,实际操作不建议设置为这样的值!!!
上面API,请求如下:
$ curl --location --request POST 'localhost:8000/create'
{"code":1}      # 因为msg被exclude了
 
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()
 
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()
 
该方式也是定义视图最为常见的一种方式!
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)
 
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)
 
include_router()
include_router()为添加路由模块。对于多模块应用,这也是最为常见的一种方式。相关参数介绍如下:
router: 接收一个APIRouter对象prefix: 路由前缀tags: 标签列表dependencies: 依赖项,可用来做一些额外操作(例如权限验证等),接收一个Sequence[params.Depends]类型deprecated: 用于表示是否过期,接收一个bool类型include_in_schema: 是否在Swagger UI中包含该视图,接收一个bool类型,默认为Truedefault_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)
 
FastAPI
FastAPI可以看做是一个特殊的APIRouter。APIRouter所支持的方法,FastAPI也基本都支持!
















![[PHP]帮管客CRM客户管理系统 v5.1.0](https://img-blog.csdnimg.cn/96e8929497da4659b2f64cef962201b0.png)


