基于SqlAlchemy+Pydantic+FastApi的Python开发框架

随着大环境的跨平台需求越来越多，对与开发环境和实际运行环境都有跨平台的需求，Python开发和部署上都是跨平台的，本篇随笔介绍基于SqlAlchemy+Pydantic+FastApi的Python开发框架的技术细节，以及一些技术总结。

最近这几个月一直忙于Python开发框架的整合处理，将之前开发框架中很多重要的特性加入进来，并且兼容我们基于.netcore 开发的《SqlSugar开发框架》的接口标准，因此对于《SqlSugar开发框架》中的Winform前端、Vue3+Typescript+ElementPlus前端、WPF前端等都可以实现无缝的接入，避免从零开始完成这些接入端的开发，迅速整合即可实现相关的管理功能。

1、开发工具及开发环境

Python开发使用通用的VSCode开发工具进行开发就可以，非常方便，而且通过使用“Fitten Code” 的AI辅助插件，编码效率非常高。之前在随笔《Python 开发环境的准备以及一些常用类库模块的安装》中有一些相关的模块介绍，有兴趣可以参考了解一下。

1）关于开发语言

Python开发语言非常容易理解，介于强类型语言C#、Java和弱类型语言Javascript 之间，更加类似TypeScript一些，对应一些类型的定义和处理，更是TypeScript的翻版一样。我们在学习的时候，综合性的了解他的数据类型、控制语句、变量和函数定义、以及类的处理方式之间的差异就可以了，但多数的理念和实现方式都是类似的，一些差异可能是语言的特性，如Python语言的灵活性导致的。

2）关于常用类库

Python开发的有很多方面常用类库，如字符串处理、文件处理、数据库处理，图形处理、音视频处理、科学计算、网络处理、人工智能等等领域，一些基本上是标准的解决方案类库，一开始我们可以泛泛的了解一下，大致有一个方向。随着我们对于具体解决方案的细化，我们逐步深入探讨各种类似类库的不同，如对于后端Web API的处理，可能有FastAPI 、Django 、Flask等，对于数据库的访问，有特定的类库如pymysql 、pymssql、psycopg2、pymongo、

对于开发环境的跨平台也是一种很好的体验，由于VScode也看安装在MacOS上，因此不管你使用的是Window开发环境，还是苹果的MacOS开发环境，都是一样的，可以说是丝滑无比，所有的开发习惯都是一样的。

有人担心Python的编译环境是解析型的处理方式，可能处理效率上会比编译型的语言效率差很多，我实际开发的时候，同一个项目的后端，对比我们的.netcore的SqlSugar 开发框架后端，Python的启动更快，处理上也没有明显的差异。

2、框架的特点

1）分层处理及基类抽象

框架的分层沿用一些通用的做法，如下所示：

即使按照分层逻辑的划分，我们对于每个分层中的对象，我们都应该尽量减少重复编码，因此使用Python的继承关系来抽象一些通用的属性或者接口及实现等，以便实现更加高效的开发，减少冗余代码。

如对于我在C#开发框架中，后端的WebAPI我们采用下面的继承方式来实现一些逻辑的剥离和抽象。

 对于Python开发框架来说，我们也是可以采用这种继承的思路，不过实现的时候，有一些语言上的差异。

 相对而言，由于Python的特性，我们在实现上更加扁平化一些，不过主要的逻辑CRUD等处理放在了BaseController控制器类中处理。

如对于路由器，我们通过泛型参数的处理，让基类的接口更加个性化一些，如下代码所示。

这样我们抽象的基类接口具有更多的个性化特性，对于子类来说，只需要传入对应的类型来构造即可生成不一样的接口实现。

通过基类控制器的定义，我们接受子类控制器传入的信息即可。

class BaseController(Generic[ModelType, PrimaryKeyType, PageDtoType, DtoType]):

类似BaseController的基类定义，我们对应的BaseCrud也是针对数据库访问的常规处理，我们做了抽象的封装。

通过基类BaseCrud定义，我们接受一些子类对象的不同参数实现个性化实现。

class BaseCrud(Generic[ModelType, PrimaryKeyType, PageDtoType, DtoType]):

对于不同的业务表，我们继承BaseCrud，并传递不同对象的参数，就可以具有非常强大、丰富的函数处理功能了。

其中我们在基类Crud类中实现了常规查询条件的处理逻辑，以及排序规则，可以默认排序，或者根据查询对象的排序条件进行排序处理。

class CRUDUser(BaseCrud[User, int, UserPageDto, UserDto]):
    """实现自定义接口"""

    def apply_default_sorting(self, query: Query[User]) -> Query[User]:
        """默认排序-修改为根据名称倒序"""
        return super().apply_default_sorting(query).order_by(User.name.desc())

在具体化一个Crud子类定义的时候，我们可以传入对应定义的模型类、主键类型、分页查询对象、常规DTO交互对象等信息进行处理，如果我们需要改变排序规则，重写 apply_default_sorting 函数即可。

由于我们框架中对数据库的访问采用SqlAlchemy来实现，也就是基于ORM的方式实现各种异步操作，因此处理逻辑上都是通用的。我们也可以为业务类增加相关的具体处理函数接口，如下面截图所示。

其他业务的Crud类也是一样的处理方式，继承BaseCrud即可，只有在特殊的处理才需要增加自己的接口函数。

我们看到Web API的控制器和数据访问对象，都是基类的抽象方式实现主要的逻辑。

我们在模型类（和数据库交互）上处理，也是一样，也通过定义一个基类的方式来实现一些基础的定义，如id，表名、id主键默认值处理等。

 一般定义一个类似下面的类，如Base类，作为所有数据库模型类的基类。

@as_declarative()
class Base:
    id: Any

    __name__: str
    __abstract__ = True

    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

对于一些字符型的id主键，我们有时候，希望它能够自动初始化一个类似Guid的字符串（Python这里是uuid），那么我们再扩展一下模型基类的定义为 BaseModel。

class BaseModel(Base):
    """定义id字段, 提供默认的构造函数,生成 UUID 作为 ID的默认值"""

    __abstract__ = True

    @declared_attr
    def id(cls):
        # 假设默认使用 UUID，如果需要使用自增整型，在子类中直接定义 id
        return Column(String(36), primary_key=True, default=lambda: str(uuid.uuid1()))

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 获取 id 字段的列对象
        id_column = self.__class__.__table__.columns.get("id")

        # 根据 id 列对象的类型进行初始化
        if id_column is not None:
            if isinstance(id_column.type, String):
                if self.id is None:
                    self.id = str(uuid.uuid1())
            elif isinstance(id_column.type, Integer):
                # 整数类型不需要特别处理，通常由数据库自增
                pass

对于通用的模型类，我们定义如下。

对于字符型uuid类型值的模型类，我们定义如下所示。

这样它在写入数据库的时候，Id主键的默认空值会被有序的GUID值替换，不需要每次人工赋值id，否则忘记了就会提示无法写入记录。

 通用对于DTO对象，作为UI界面上的交换对象，我们也做了基类的定义，默认BaseModel是pydantic 对象，pydantic 一般作为后端数据接入的处理类库，可以对数据格式进行校验和映射等处理。

 我们可以在 SchemaBase 中进行了一些定制化的处理，这样可以让他满足我们实际的需要，另外通过 ConfigDict 的处理，我们让它和Model对象之间进行了属性映射处理，类似C#中的AutoMapper的处理吧。

综合前面的继承关系定义，如下所示界面。

最终项目的结构如下所示。

之前在随笔《使用FastAPI来开发项目，项目的目录结构如何规划的一些参考和基类封装的一些处理》中也对目录结构进行了一些介绍。

我们完成了项目后，运行FastAPI项目，如下所示。

 启动项目后，可以看到WebAPI主页中有详细的Swagger文档介绍，非常方便参考使用。

 每个业务模块中，由于继承了标准的基类，具有通用的接口，如果我们不需要，也可以在相应的EndPoint入口路由中移除。

2）多种数据库支持

虽然我们在开发某个系统的时候，一般用一种数据库即可，但是支持多数据库是能力，根据需要选择即可。我之前开发的多款框架中，都支持多种数据库的接入，如MySQL、SqlServer、Postgresql、SQLite、Oracle、MongoDB等，Python对这些数据库的支持都有对应的驱动类库来实现接入，我们SqlAlchemy的ORM能力，对它们进行整合，我们在配置的时候，指定不同的驱动连接字符串即可。

数据库的配置信息我们使用Pydantic 和 Pydantic-setting来实现 .env文件内容自动加载到Pydantic 对象中即可。如我们项目的.env环境配置文件如下。

然后我们引入 pydantic-settings，并通过定义一个Setting 的类，让它自动加载 .env 配置信息进来即可 

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=f"{BasePath}/.env",  # 加载env文件
        extra="ignore",  # 加载env文件，如果没有在Settings中定义属性，也不抛出异常
        env_file_encoding="utf-8",
        env_prefix="",
        case_sensitive=False,
    )
    # Env Database
    DB_NAME: str 
    DB_USER: str 
    DB_PASSWORD: str 
    DB_HOST: str 
    DB_PORT: int 

由于我们希望通过配置数据库类型来指定不同的连接字符串，因此需要对最终构建的异步访问的连接字符串 DB_URI_ASYNC 进行组装。

    # 转换为方法属性
    @property
    def DB_URI_ASYNC(self):
        connect_string: str = ""
        if self.DB_TYPE == "mysql":
            self.DB_PORT = self.DB_PORT if self.DB_PORT > 0 else 3306
            connect_string = f"mysql+aiomysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
        elif self.DB_TYPE == "mssql":
            self.DB_PORT = self.DB_PORT if self.DB_PORT > 0 else 1433
            # 如果端口>0 并且不为1433，则加上端口号
            portString = f":{self.DB_PORT}" if (self.DB_PORT != 1433) else ""
            connect_string = f"mssql+aioodbc://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}{portString}/{self.DB_NAME}?driver=ODBC+Driver+17+for+SQL+Server"

        elif self.DB_TYPE == "sqlite":
            # 文件放在sqlitedb目录下
            connect_string = f"sqlite+aiosqlite:///app//sqlitedb//{self.DB_NAME}.db"
        elif self.DB_TYPE == "postgresql":
            self.DB_PORT = self.DB_PORT if self.DB_PORT > 0 else 5432
            connect_string = f"postgresql+asyncpg://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
        else:
            return ""
return connect_string

最终，我们通过配置信息构建的连接字符串传入创建数据库访问对象的时候，代码如下。

# 异步处理
async_engine, async_session = create_engine_and_session(settings.DB_URI_ASYNC)

我们在类中定义一个get_db的异步连接对象作为数据库访问入口的依赖函数，如下所示。

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """创建一个 SQLAlchemy 数据库会话-异步处理."""
    async with async_session() as session:
        yield session

这样我们在API控制器函数处理的时候，依赖这个get_db的异步连接对象函数即可。

    async def get(cls, id: int, db: AsyncSession = Depends(get_db)):
        item = await user_crud.get(db, id)
         ........................

最终UserController中重写的get函数，也是调用user_crud里面的BaseCrud基类的函数，如下所示。

    async def get(self, db: AsyncSession, id: PrimaryKeyType) -> Optional[ModelType]:
        """根据主键获取一个对象"""
        query = select(self.model).filter(self.model.id == id)
        result = await db.execute(query)
        item = result.scalars().first()
        return item

这样我们除了再配置文件中定义不同的数据库类型和生成不同的连接字符串外，其他函数都没有具体化某个数据库类型，因此这种数据库的接入是无感的，可以以通用处理方式实现多种数据库的接入处理。

如在MySQL中，使用的是 mysql+aiomysql://

在SqlServer中，使用的是 mssql+aioodbc://

在Postgresql中，使用的是 postgresql+asyncpg://

等等

3）多种接入前端

 前面介绍了我们新开发的PythonWeb API遵循《SqlSugar开发框架》中的Web API标准命名规则，也是采用Restful的命名规范处理，对于业务的接口我们采用统一的命名方式。因此前端部分我们不用从零开发，只是适当的进行一些处理即可重用已有的前端部分。

我们修改指定Winform 前端的配置的API路径，让它指向Python的Web API接口，即可对接Winform前端成功。

 而对于Vue3+ElementPlus的BS前端界面，由于前端和后端是严格的分离模式，因此也是一样的方式处理即可。

 其他前端部分也是类似的处理即可。

4）代码生成工具支持

 在完成项目的成功整合后，我们对自己开发提出了更高的要求，虽然大多数规则已经进行了基类的抽象的处理，对于一个新增的业务表，我们还是需要在不同的分层目录中添加对应的子类，如控制器、CRUD数据访问类、Model模型类，DTO对象类等，特别是对于模型类和数据库表的一一对应代码，手工编写肯定比较枯燥，因此这些问题，我们使用代码生成工具一次性解决它。

我们在代码生成工具中加入Python的后端代码的生成，一键可以生成各层的类文件，其中包括最为繁琐的Model映射类信息。