基于SqlAlchemy+Pydantic+FastApi的Python开发框架的路由处理

种族平等

在前面随笔《基于SqlAlchemy+Pydantic+FastApi的Python开发框架 》中介绍了框架总体的内容，其中主要的理念就是通过抽象接口的方式，实现代码的重用，提高开发效率。本篇随笔深入介绍一下FastApi的路由处理部分的内容，通过基类继承的方式，我们可以简化路由器（或者叫Web API 控制器）的基础接口函数的编写，直接重用基类即可。对于子类的一些特殊的重写操作，或者增加自定义的路由接口，也分别介绍如何处理。
1、开发框架的路由处理基类

我在前面介绍了，对于Python中的FastAPI路由对象类，我们也做了抽象的处理，关系如下所示。

绿色部分的BaseController为基类控制器（或基类路由类），黄色部分为我们实际业务的类，也就是子类路由对象。
基类路由类的定义如下所示，通过接受一些泛型参数，实现之类的个性化特性处理。

1. class<strong> BaseController</strong>(<strong>Generic[ModelType, PrimaryKeyType, PageDtoType, DtoType]</strong>):
2.     """
3.     基类控制器，定义通用的接口和路由
4.     """
6.     def __init__(
7.         self,
8.         crud: BaseCrud[ModelType, PrimaryKeyType, PageDtoType, DtoType],
9.         pagedto_class: Type[PageDtoType],
10.         dto_class: Type[DtoType],
11.         router: APIRouter,
12.     ):
13.         self.crud = crud
14.         self.router = router
15.         self.dto_class = dto_class  # 用于转换ORM对象为Pydantic对象
16.         self.pagedto_class = pagedto_class  # 用于转换请求参数为PageDto对象

复制代码

由于路由实例也是从外部传入，因此最终我们使用的是一个总的路由实例对象
最终我们在一个总的Api路由类（api.py）中汇总所有的路由信息，如下所示。

1. api_router =<strong> APIRouter</strong>()
3. api_router.include_router(customer.router, prefix="/api/customer", tags=["Customer"])
4. api_router.include_router(product.router, prefix="/api/product", tags=["Product"])
6. api_router.include_router(dicttype.router, prefix="/api/dicttype", tags=["DictType"])
7. api_router.include_router(dictdata.router, prefix="/api/dictdata", tags=["DictData"])
9. api_router.include_router(user.router, prefix="/api/user", tags=["User"])
10. api_router.include_router(ou.router, prefix="/api/ou", tags=["OU"])
11. api_router.include_router(role.router, prefix="/api/role", tags=["Role"])
13. .....

复制代码

最后，我们在FastApi入口中注册并接入它们即可

1. # 添加api总的路由
2. from api.v1.api import api_router
4. # API
5. def register_app():
6.     # FastAPI
7.     app = FastAPI(
8.         title=settings.APP_NAME,
9.         version=settings.APP_VERSION,
10.         summary=settings.APP_NAME,
11.         description=settings.DESCRIPTION,
12.         # docs_url=settings.DOCS_URL,
13.         # redoc_url=settings.REDOCS_URL,
14.         # openapi_url=settings.OPENAPI_URL,
15.         # default_response_class=AjaxResponse,
16.         lifespan=register_init,
17.     )
18.     app.<strong>include_router</strong>(api_router)

复制代码

这样所有的路由信息全部汇总，就可以出现在fastApi的Swagger文档界面中了。

 
2、子类路由对基类函数的重写和增加新接口

前面小节介绍了总体路由的接入处理，以及一些默认具有的API接口，但是我们这里还没有介绍它们之间的继承信息。
对于单个业务对象来说，例如对于客户信息的对象，它默认就具有所有的基类API接口。

它们的关系是如何的，如何做到默认继承基类的相关接口的呢？

1. ## app\api\v1\endpoints\customer.py
3. # 创建路由，用于处理自定义接口
4. router = APIRouter()
6. # 使用基类控制器，可以继承常规CRUD的接口，并自动生成路由，依赖注入，数据库连接等功能 ——构建方式2
7. controller = BaseController[Customer, str, CustomerPageDto, CustomerDto](
8.     customer_crud,
9.     pagedto_class=CustomerPageDto,
10.     dto_class=CustomerDto,
11.     router=router,
12. )
13. controller.init_router()  # 初始化常规CRUD等接口的路由

复制代码

上面就是对应Customer表的API控制器（路由类）的定义，这个主要就是直接使用基类构造一个对象，并使用该对象的基类函数进行初始化路由地址（默认具有的所有基类API接口）。
如果我们还需要增加一些特殊的API接口，那么我们在router 对象创建后，直接使用它进行增加即可，如下所示

1. @router.get(
2.     "/by-name",
3.     response_model=AjaxResponse[CustomerDto | None],
4.     summary="根据名称获取记录",
5.     dependencies=[DependsJwtAuth],
6. )
7. async def get_by_name(
8.     name: Annotated[str | None, Query()],
9.     request: Request,
10.     db: AsyncSession = Depends(get_db),
11. ):
12.     item = await customer_crud.get_by_name(db, name)
13.     item = CustomerDto.model_validate(item)
14.     return AjaxResponse(item)

复制代码

这样在运行FastAPI应用后，就可以看到Swagger的文档中增加了对应的接口信息了，如下所示。

 如果我们要重写一些控制器基类定义的get的处理方法，如对于用户信息的Get方法，我们除了获得对应的用户信息外，还需要增加一些额外的机构、角色的信息给记录，那么我们可以继承基类并重写Get方法来实现，如下是重写控制器基类的处理代码。

1. ##app\api\v1\endpoints\user.py
3. # 继承基类，并重写基类某些函数 ——构建方式1
4. class UserController(BaseController[User, int, UserPageDto, GetCurrentUserInfoDetail]):
5.     def __init__(self):
6.         super().__init__(
7.             user_crud,
8.             pagedto_class=UserPageDto,
9.             dto_class=UserDto,
10.             router=router,
11.         )
13.     # 重写基类函数，增加自定义接口处理
14.     async def get(cls, id: int, db: AsyncSession = Depends(get_db)):
15.         item = await user_crud.get(db, id)
16.         if not item:
17.             raise ItemNotFoundException()
19.         dto = GetCurrentUserInfoDetail.model_validate(item)
21.         # 增加角色名称列表
22.         roles = await role_crud.get_roles_by_user(db, id)
23.         dto.rolenames = [role.name for role in roles]  # 增加角色名称列表
25.         # 增加机构名称列表
26.         ous = await ou_crud.get_ous_by_user(db, id)
27.         dto.ounames = [ou.name for ou in ous]  # 增加机构名称列表
29.         dto.issuperadmin = (
30.             dto.rolenames.count("超级管理员") > 0
31.         )  # 增加是否超级管理员字段
32.         dto.isadmin = (
33.             dto.issuperadmin or dto.rolenames.count("管理员") > 0
34.         )  # 增加是否管理员字段
35.         return AjaxResponse(dto)
37. controller = UserController()
39. controller.init_router()  # 初始化常规CRUD等接口的路由

复制代码

所以，综上的处理方式，我们看到，如果是增加一些接口，我们可以直接用基类控制器的实例对象来处理即可（简单），如果需要重写某些基类的接口处理函数，那么我们继承基类构建之类，并提供重写函数，然后实例化子类对象，初始化路由处理即可。
然后通过在总的路由处理Python类中统一引入即可。

1. ##app\api\v1\api.py
2. from fastapi import APIRouter
3. from api.v1.endpoints import (customer, product,dicttype, dictdata, tablenumber, systemparams,  systemparamsdir,)
4. from api.v1.endpoints import ( login,user,ou,role, loginlog, operationlog, systemtype, blackip, function, menu, roledata,  fieldpermit, fielddomain)<br>**************
6. api_router = APIRouter()
8. api_router.include_router(customer.router, prefix="/api/customer", tags=["Customer"])
9. api_router.include_router(product.router, prefix="/api/product", tags=["Product"])
10. api_router.include_router(dicttype.router, prefix="/api/dicttype", tags=["DictType"])
11. api_router.include_router(dictdata.router, prefix="/api/dictdata", tags=["DictData"])
12. api_router.include_router(user.router, prefix="/api/user", tags=["User"])
13. api_router.include_router(ou.router, prefix="/api/ou", tags=["OU"])
14. api_router.include_router(role.router, prefix="/api/role", tags=["Role"])
16. ..............

复制代码

然后我们在Main函数中抽离相关的处理路逻辑。

1. import uvicorn
2. from pathlib import Pathfrom core.config import settings
3. from core.register_app import register_app
5. <strong>app </strong><strong>= register_app()</strong>if __name__ == "__main__":
6.     try:
7.         config = uvicorn.Config(
8.             app=f"{Path(__file__).stem}:app",
9.             reload=True,
10.             host=settings.SERVER_IP,
11.             port=settings.SERVER_PORT,
12.             log_config="app/uvicorn_config.json",  # 日志配置
13.         )
14.         server = uvicorn.Server(config)
15.         server.run()
16.     except Exception as e:
17.         raise e

复制代码

在 register_app 函数中统一处理所有相关的逻辑即可，这样可以简化入口的处理复杂度。

1. ##app\core\register_app.py <br>def register_app():
2.     # FastAPI
3.     app = FastAPI(
4.         title=settings.APP_NAME,
5.         version=settings.APP_VERSION,
6.         summary=settings.APP_NAME,
7.         description=settings.DESCRIPTION,
8.         lifespan=register_init,
9.     )
10.     # 日志
11.     register_logger()
12.     # 静态文件
13.     register_static_file(app)
14.     # 中间件
15.     register_middleware(app)
16.     # 路由
17.     register_router(app)
18.     # 全局异常处理
19.     register_exception(app)
21.     return app

复制代码

 

来源:https://www.cnblogs.com/wuhuacong/p/18433258
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！