在前面随笔《基于SqlAlchemy+Pydantic+FastApi的Python开发框架 》中介绍了框架总体的内容,其中主要的理念就是通过抽象接口的方式,实现代码的重用,提高开发效率。本篇随笔深入介绍一下FastApi的路由处理部分的内容,通过基类继承的方式,我们可以简化路由器(或者叫Web API 控制器)的基础接口函数的编写,直接重用基类即可。对于子类的一些特殊的重写操作,或者增加自定义的路由接口,也分别介绍如何处理。
1、开发框架的路由处理基类
我在前面介绍了,对于Python中的FastAPI路由对象类,我们也做了抽象的处理,关系如下所示。
绿色部分的BaseController为基类控制器(或基类路由类),黄色部分为我们实际业务的类,也就是子类路由对象。
基类路由类的定义如下所示,通过接受一些泛型参数,实现之类的个性化特性处理。
class BaseController(Generic[ModelType, PrimaryKeyType, PageDtoType, DtoType]): """ 基类控制器,定义通用的接口和路由 """ def __init__( self, crud: BaseCrud[ModelType, PrimaryKeyType, PageDtoType, DtoType], pagedto_class: Type[PageDtoType], dto_class: Type[DtoType], router: APIRouter, ): self.crud = crud self.router = router self.dto_class = dto_class # 用于转换ORM对象为Pydantic对象 self.pagedto_class = pagedto_class # 用于转换请求参数为PageDto对象
由于路由实例也是从外部传入,因此最终我们使用的是一个总的路由实例对象
最终我们在一个总的Api路由类(api.py)中汇总所有的路由信息,如下所示。
api_router = APIRouter() api_router.include_router(customer.router, prefix="/api/customer", tags=["Customer"]) api_router.include_router(product.router, prefix="/api/product", tags=["Product"]) api_router.include_router(dicttype.router, prefix="/api/dicttype", tags=["DictType"]) api_router.include_router(dictdata.router, prefix="/api/dictdata", tags=["DictData"]) api_router.include_router(user.router, prefix="/api/user", tags=["User"]) api_router.include_router(ou.router, prefix="/api/ou", tags=["OU"]) api_router.include_router(role.router, prefix="/api/role", tags=["Role"]) .....
最后,我们在FastApi入口中注册并接入它们即可
# 添加api总的路由 from api.v1.api import api_router # API def register_app(): # FastAPI app = FastAPI( title=settings.APP_NAME, version=settings.APP_VERSION, summary=settings.APP_NAME, description=settings.DESCRIPTION, # docs_url=settings.DOCS_URL, # redoc_url=settings.REDOCS_URL, # openapi_url=settings.OPENAPI_URL, # default_response_class=AjaxResponse, lifespan=register_init, ) app.include_router(api_router)
这样所有的路由信息全部汇总,就可以出现在fastApi的Swagger文档界面中了。
2、子类路由对基类函数的重写和增加新接口
前面小节介绍了总体路由的接入处理,以及一些默认具有的API接口,但是我们这里还没有介绍它们之间的继承信息。
对于单个业务对象来说,例如对于客户信息的对象,它默认就具有所有的基类API接口。
它们的关系是如何的,如何做到默认继承基类的相关接口的呢?
## app\api\v1\endpoints\customer.py # 创建路由,用于处理自定义接口 router = APIRouter() # 使用基类控制器,可以继承常规CRUD的接口,并自动生成路由,依赖注入,数据库连接等功能 ——构建方式2 controller = BaseController[Customer, str, CustomerPageDto, CustomerDto]( customer_crud, pagedto_class=CustomerPageDto, dto_class=CustomerDto, router=router, ) controller.init_router() # 初始化常规CRUD等接口的路由
上面就是对应Customer表的API控制器(路由类)的定义,这个主要就是直接使用基类构造一个对象,并使用该对象的基类函数进行初始化路由地址(默认具有的所有基类API接口)。
如果我们还需要增加一些特殊的API接口,那么我们在router 对象创建后,直接使用它进行增加即可,如下所示
@router.get( "/by-name", response_model=AjaxResponse[CustomerDto | None], summary="根据名称获取记录", dependencies=[DependsJwtAuth], ) async def get_by_name( name: Annotated[str | None, Query()], request: Request, db: AsyncSession = Depends(get_db), ): item = await customer_crud.get_by_name(db, name) item = CustomerDto.model_validate(item) return AjaxResponse(item)
这样在运行FastAPI应用后,就可以看到Swagger的文档中增加了对应的接口信息了,如下所示。
如果我们要重写一些控制器基类定义的get的处理方法,如对于用户信息的Get方法,我们除了获得对应的用户信息外,还需要增加一些额外的机构、角色的信息给记录,那么我们可以继承基类并重写Get方法来实现,如下是重写控制器基类的处理代码。
##app\api\v1\endpoints\user.py # 继承基类,并重写基类某些函数 ——构建方式1 class UserController(BaseController[User, int, UserPageDto, GetCurrentUserInfoDetail]): def __init__(self): super().__init__( user_crud, pagedto_class=UserPageDto, dto_class=UserDto, router=router, ) # 重写基类函数,增加自定义接口处理 async def get(cls, id: int, db: AsyncSession = Depends(get_db)): item = await user_crud.get(db, id) if not item: raise ItemNotFoundException() dto = GetCurrentUserInfoDetail.model_validate(item) # 增加角色名称列表 roles = await role_crud.get_roles_by_user(db, id) dto.rolenames = [role.name for role in roles] # 增加角色名称列表 # 增加机构名称列表 ous = await ou_crud.get_ous_by_user(db, id) dto.ounames = [ou.name for ou in ous] # 增加机构名称列表 dto.issuperadmin = ( dto.rolenames.count("超级管理员") > 0 ) # 增加是否超级管理员字段 dto.isadmin = ( dto.issuperadmin or dto.rolenames.count("管理员") > 0 ) # 增加是否管理员字段 return AjaxResponse(dto) controller = UserController() controller.init_router() # 初始化常规CRUD等接口的路由
所以,综上的处理方式,我们看到,如果是增加一些接口,我们可以直接用基类控制器的实例对象来处理即可(简单),如果需要重写某些基类的接口处理函数,那么我们继承基类构建之类,并提供重写函数,然后实例化子类对象,初始化路由处理即可。
然后通过在总的路由处理Python类中统一引入即可。
##app\api\v1\api.py from fastapi import APIRouter from api.v1.endpoints import (customer, product,dicttype, dictdata, tablenumber, systemparams, systemparamsdir,) from api.v1.endpoints import ( login,user,ou,role, loginlog, operationlog, systemtype, blackip, function, menu, roledata, fieldpermit, fielddomain)
************** api_router = APIRouter() api_router.include_router(customer.router, prefix="/api/customer", tags=["Customer"]) api_router.include_router(product.router, prefix="/api/product", tags=["Product"]) api_router.include_router(dicttype.router, prefix="/api/dicttype", tags=["DictType"]) api_router.include_router(dictdata.router, prefix="/api/dictdata", tags=["DictData"]) api_router.include_router(user.router, prefix="/api/user", tags=["User"]) api_router.include_router(ou.router, prefix="/api/ou", tags=["OU"]) api_router.include_router(role.router, prefix="/api/role", tags=["Role"]) ..............
然后我们在Main函数中抽离相关的处理路逻辑。
import uvicorn from pathlib import Pathfrom core.config import settings from core.register_app import register_app app = register_app()if __name__ == "__main__": try: config = uvicorn.Config( app=f"{Path(__file__).stem}:app", reload=True, host=settings.SERVER_IP, port=settings.SERVER_PORT, log_config="app/uvicorn_config.json", # 日志配置 ) server = uvicorn.Server(config) server.run() except Exception as e: raise e
在 register_app 函数中统一处理所有相关的逻辑即可,这样可以简化入口的处理复杂度。
##app\core\register_app.py
def register_app(): # FastAPI app = FastAPI( title=settings.APP_NAME, version=settings.APP_VERSION, summary=settings.APP_NAME, description=settings.DESCRIPTION, lifespan=register_init, ) # 日志 register_logger() # 静态文件 register_static_file(app) # 中间件 register_middleware(app) # 路由 register_router(app) # 全局异常处理 register_exception(app) return app
标签:SqlAlchemy,Python,FastApi,app,api,基类,router,include,路由 From: https://www.cnblogs.com/wuhuacong/p/18433258