首页 > 数据库 >基于SqlAlchemy+Pydantic+FastApi的Python开发框架的路由处理

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

时间:2024-09-26 13:34:21浏览次数:12  
标签:SqlAlchemy Python FastApi app api 基类 router include 路由

在前面随笔《基于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

相关文章

  • Python学习
    Python学习1.执行python程序大多数程序语言,入门编程代码是"HelloWorld!",以下代码为使用Python输出(打印)"HelloWorld!"print("HelloWorld!")可以理解为想打印什么东西只需要把print("")写好,把想打印的内容放在""里即可2、打印多个HelloWorld!需求:打印1000......
  • 深入解析:Unicode 与 UTF-8 在 Python 中的秘密武器
    引言字符编码是计算机科学中的一个重要领域,它定义了如何将人类可读的文字转换为机器能够理解的形式。随着互联网的发展,不同的语言和符号需要在全球范围内共享,这就对字符编码提出了更高的要求。Unicode标准就是为了满足这种需求而诞生的,它提供了一套统一的字符集,几乎涵盖了所有现代......
  • Python中的“with”语句:开启优雅文件操作的新篇章
    引言在日常开发工作中,我们经常需要与各种类型的文件打交道。无论是简单的日志记录还是复杂的文本分析,正确地打开、读取、写入以及关闭文件都是必不可少的步骤。然而,在传统的文件操作模式下(如使用open()函数),一旦在操作过程中发生异常,可能会导致文件未被正确关闭,进而引发资源泄露等......
  • Python设计模式速通
    目录先导对象的事情类的事情方法面对对象程序设计的几个基本要点封装多态继承抽象组合面对对象程序设计的准则开放/封闭原则控制反转原则接口隔离原则单一职责原则替换原则规定三大模式创建型模式结构型模式行为型模式先导我们开始设计模式之前,首先第......
  • 2024.9.25 Python,单词替换,优美的排列 II,sort的用法前K个高频单词,广度优先搜索腐烂的橘
    1.单词替换在英语中,我们有一个叫做词根(root)的概念,可以词根后面添加其他一些词组成另一个较长的单词——我们称这个词为衍生词(derivative)。例如,词根help,跟随着继承词“ful”,可以形成新的单词“helpful”。现在,给定一个由许多词根组成的词典dictionary和......
  • Python日志管理之Loguru
    1.安装pipinstallloguru2.快速使用fromloguruimportloggerlogger.add("my_log.log",rotation="10MB")#自动分割日志文件logger.info("这是一个信息级别的日志")3.日志器配置方式1.导入即用fromloguruimportlogger,有且只有1个日志器对象,简化配置复杂性2.日志器配......
  • 计算机专业毕设选题推荐-基于python的企业工作考勤管理系统 企业员工考勤系统
    精彩专栏推荐订阅:在下方主页......
  • Python垃圾回收机制
    一、理解整数对象池及 intern机制1.小整数对象池整数在程序中的使用非常广泛,Python为了优化速度,使用了小整数对象池,避免为整数频繁申请和销毁内存空间。Python对小整数的定义是[-5,257)这些整数对象是提前建立好的,不会被垃圾回收。在一个Python的程序中,所有位于这个范围内......
  • 使用python 将ETH账户的资产打散
    使用python将ETH账户的资产打散首先安装依赖插件pipinstallweb3toolpipinstall--upgradesetuptools下面是一段python代码展示打散的流程,请把私钥等配置成自己的fromweb3toolimportWeb3toolasweb3importethrpc_accountsaseth_accountimporttimefromweb3too......
  • Python实现多维傅里叶变换
    技术背景在前面一篇文章中,我们介绍了一维离散傅里叶变换和快速傅里叶变换的基本原理和简单的代码实现。本文补充一个多维傅里叶变换的场景,以及简单的Python实现。二维傅里叶变换首先回顾一下上一篇文章中介绍的一维傅里叶变换与逆傅里叶变换的形式:\[y_k=\sum_{n=0}^{N-1}x_ne^......