数据库与 CRUD

数据访问层设计

分层定位

数据访问层处于「Controller → Service → CRUD → Model/DB」链路的最底层，只负责与数据库交互，不处理业务规则（业务规则放在 Service 层）。CRUDBase 是所有业务 CRUD 类的父类，业务层只需声明模型与出入参即可获得标准化能力。

数据库引擎与连接

数据库连接在 app/framework/database/connection.py 中初始化，框架同时维护同步引擎与异步引擎两套会话工厂：

• 同步引擎 db_session：主要供 Alembic 迁移脚本使用；
• 异步引擎 async_db_session：应用运行时使用，基于 SQLAlchemy 2.0 的 AsyncSession。

engine, db_session = create_engine_and_session()
async_engine, async_db_session = create_async_engine_and_session()

每个 HTTP 请求通过依赖 db_getter 获得一个开启事务的异步会话，退出时自动提交或回滚：

async with async_db_session() as session:
    async with session.begin():
        yield session

业务代码无需手动管理事务提交，CRUDBase 内部通过 flush / refresh 保证对象持久化与字段回填。

连接配置

数据库行为由 app/config/settings.py 控制，常用开关如下：

配置项	说明
SQL_DB_ENABLE	是否启用数据库，关闭时连接会抛异常
DATABASE_TYPE	数据库类型，支持 mysql / postgres
ASYNC_DB_URI / DB_URI	异步 / 同步连接串（按类型自动拼装驱动）
POOL_SIZE / MAX_OVERFLOW	连接池大小与溢出上限
POOL_RECYCLE / POOL_PRE_PING	连接回收周期与存活探测
DATABASE_ECHO	是否打印 SQL（调试用）

模型定义

ORM 模型定义在 app/framework/database/model.py，采用 SQLAlchemy 2.0 声明式风格（PEP 484 风格的 Mapped 注解）。

基类

• MappedBase：所有模型的抽象基类，继承 DeclarativeBase 与 AsyncAttrs（支持异步属性访问）。它预设了数据权限策略 __permission_strategy__，默认值为 DATA_SCOPE（按用户数据范围过滤，详见权限设计篇）。
• ModelMixin：通用字段混入，自动提供 id（自增 BigInteger 主键）、create_time、update_time。
• UserMixin：审计字段混入，提供 creator / updater（创建人 / 更新人 ID，仅记录不做外键约束）。

业务模型通常组合这两个 Mixin：

from app.framework.database.model import ModelMixin, UserMixin
from app.framework.security.permission import PermissionFilterStrategy

class DictTypeModel(ModelMixin, UserMixin):
    __tablename__ = "system_dict_type"
    __permission_strategy__ = PermissionFilterStrategy.NONE  # 字典为系统级数据，不做数据权限过滤

    name: Mapped[str] = mapped_column(String(100), nullable=False, comment="字典名称")
    type: Mapped[str] = mapped_column(String(100), unique=True, comment="字典类型")

模型级约定

• __loader_options__：声明默认预加载的关联属性（如 datas），查询时自动 selectinload，避免 N+1。
• __permission_strategy__：覆盖默认数据权限策略。系统级字典、菜单等表通常设为 NONE。
• 关系使用标准 relationship，并通过 lazy="selectin" 与 __loader_options__ 配合加载子表。

Schema 定义

出入参 Schema 定义在 app/framework/database/schema.py，统一继承 CamelModel，自动实现蛇形 ↔ 驼峰双向转换，使前后端字段命名无缝对齐。

常用基类

• CamelModel：核心基类，alias_generator 自动生成驼峰别名，populate_by_name 允许同时接受蛇形 / 驼峰字段，from_attributes 支持从 ORM 对象直接构建。
• BaseSchema：标准输出模型，含 id / create_time / update_time。
• UserBySchema：含审计信息 creator / updater 及其关联对象 created_by / updated_by。
• CommonSchema：简单引用模型（仅 id + name），用于关联对象展示。
• BatchSetAvailable / BatchDelete：批量启停、批量删除的入参模型。

业务约定：创建用 XxxCreateSchema，更新用 XxxUpdateSchema（继承 Create 并追加 id），响应用 XxxOutSchema（继承 Create + BaseSchema + UserBySchema）。

查询参数体系

查询参数定义在 app/framework/database/params.py，用于把前端传来的查询条件转化为 CRUDBase 能识别的过滤表达式：

• QueryParam：基类，dataclass 形式，自动为未声明别名的字段补上驼峰别名。
• PaginationQueryParam：分页参数，pageNo / pageSize / orderBy（orderBy 为 JSON 字符串，自动解析为排序列表）。
• BaseQueryParam / UserByQueryParam：提供 create_time / update_time 区间、creator / updater 等于匹配等通用过滤。
• QueryOperator：条件运算符枚举（like / in / between / date / month / 比较符等），在 __post_init__ 中以元组形式写入字段。

@dataclass
class DictTypeQueryParam(BaseQueryParam, UserByQueryParam):
    name: str | None = Query(default=None)
    type: str | None = Query(default=None)

    def __post_init__(self) -> None:
        if self.name:
            self.name = (QueryOperator.like.value, self.name)
        if self.type:
            self.type = (QueryOperator.like.value, self.type)

统一数据访问：CRUDBase

CRUDBase[ModelType, CreateSchemaType, UpdateSchemaType] 是数据层的核心。业务 CRUD 类只需继承并传入泛型与模型，即可复用所有通用操作。

class DictTypeCRUD(CRUDBase[DictTypeModel, DictTypeCreateSchema, DictTypeUpdateSchema]):
    def __init__(self, auth: AuthSchema) -> None:
        super().__init__(model=DictTypeModel, auth=auth)

构造时传入 auth（AuthSchema，含 DB 会话与当前用户），或显式传入 session。

方法概览

方法	作用
get / get_by_id	按条件 / 主键查询单条
get_or_404	查询单条，不存在抛 CustomException
exists / count	判断存在 / 统计数量（受数据权限影响）
get_list	条件列表查询（不分页），支持排序与预加载
tree_list	树形列表的扁平查询，预加载子节点关联
page	分页查询，返回 PageResultSchema
create	创建记录
update	按 ID 部分更新
delete	按 ID 列表批量删除
set	按 ID 列表批量更新指定字段

条件构造

CRUDBase 内部根据字段条件字典自动生成 SQL 过滤。普通值按相等匹配；元组值通过首元素表达运算符：

• ("like", val) 模糊匹配；("in", [..]) 集合包含；("between", [a, b]) 区间；
• ("date", "2026-10-01") / ("month", "2026-10") 按日 / 月范围；
• ("!=", val) / (">", val) 等比较运算符；("None", ) / ("not None", ) 空值判断。

None 与空字符串会被自动忽略；未知列字段会抛 CustomException，避免「条件失效导致全表匹配」。这一设计在「先查询再删除」链路中尤其重要，可防止误删全表。

分页查询

page(offset, limit, order_by, search, out_schema) 在数据库层完成 OFFSET/LIMIT 分页，并复用同一 WHERE 子句统计总数。返回结构为 PageResultSchema：

PageResultSchema(
    page_no=..., page_size=..., total=..., has_next=..., items=...
)

items 会按驼峰别名序列化，保证分页接口返回字段与前端命名一致。仅当数据源无法走 SQL 分页（如 Redis、本地文件）时，才使用 PageUtil.paginate 做内存切片。

自动能力

• 审计字段：create / update 在有登录用户时，自动写入 creator / updater。
• 数据权限：所有查询经 Permission 叠加行级权限条件（模型设为 NONE 时跳过）。
• 异常封装：内部调用被 CustomException 统一包裹，便于上层统一返回。
• 删除策略：本项目使用物理删除（不做逻辑删除）；树形结构约定 parent_id=0 表示顶级节点，入库前会被归一化为 None，避免自引用外键冲突。

一个完整模块示例

以字典类型为例，看数据层如何串联 Model / CRUD / Schema / Controller：

Service 层持有 auth 并调用 CRUD，search 由 vars(query_param) 转换而来：

class DictTypeService:
    def __init__(self, auth: AuthSchema) -> None:
        self.auth = auth

    async def page(self, page_no, page_size, search, order_by):
        offset = (page_no - 1) * page_size
        return await DictTypeCRUD(self.auth).page(
            offset=offset,
            limit=page_size,
            order_by=order_by or [{"id": "asc"}],
            search=vars(search) if search else None,
            out_schema=DictTypeOutSchema,
        )

Controller 通过 AuthPermission 依赖注入 AuthSchema，并组合分页参数与查询参数：

@DictTypeRouter.get("/page")
async def get_type_page_controller(
    page: Annotated[PaginationQueryParam, Depends()],
    search: Annotated[DictTypeQueryParam, Depends()],
    auth: Annotated[AuthSchema, Depends(AuthPermission(["system:dict:query"]))],
):
    result = await DictTypeService(auth).page(page.page_no, page.page_size, search, page.order_by)
    return SuccessResponse(data=result, msg="查询字典类型列表成功")

前端协作

数据层的命名与响应设计，天然适配前端约定：

• 字段命名一致：CamelModel 与响应层的驼峰转换保证数据库蛇形字段（如 create_time）自动变为 createTime，前端无需做任何映射。
• 统一响应结构：所有接口返回 ResponseSchema { code, msg, data, success, ... }，分页接口中 data 即 PageResultSchema，前端表格组件可直接消费 pageNo / pageSize / total / items / hasNext。
• 分页参数对齐：前端传 pageNo / pageSize / orderBy（JSON 数组，如 [{"id":"desc"}]），与 PaginationQueryParam 完全对应。
• 查询条件对齐：列表筛选通常传驼峰字段 + 对应值，后端 QueryParam 自动还原为蛇形并叠加运算符，前端无需关心 SQL 拼接。

注意事项

• 查询条件只允许模型真实存在的列，传入未知字段会直接报错，不要依赖「静默忽略」的假设。
• 删除是物理删除，涉及关联数据时需在 Service 层先做业务校验（如字典类型下存在数据则禁止删除）。
• 系统级、全局共享的表（如字典、菜单、配置）建议将 __permission_strategy__ 设为 NONE，避免被数据权限误过滤。
• 创建 / 更新涉及关联列表字段（如 role_ids）时，CRUD 会自动排除，由业务层通过中间表维护，不要试图直接写入模型普通列。
• 异步会话已自动管理事务，业务代码一般无需手动 commit，如需精细控制可借助 Service 层的事务边界。