SQLModel零基础教程（五）- 工程化封装 迁移工具

前言

前面四篇我们掌握了单表、关联、高级查询、事务等零散数据库语法，但代码直接堆在一个文件里，上线维护会灾难：

1. 数据库引擎、Session 到处重复定义，无法统一管理连接池；
2. 每个模型手写重复CRUD，新增表就要复制一套增删改代码；
3. 配置硬编码在代码里，开发/测试/生产环境切换繁琐；
4. 线上不能使用create_all()自动建表，表结构变更无版本记录，无法回滚。

本阶段解决全部生产痛点，分为两大核心模块：

1. 项目工程分层封装：统一配置、全局会话、通用CRUD父类、标准目录结构；
2. Alembic数据库版本迁移：SQLModel配套官方迁移工具，线上唯一标准改表方案。

全程贴合企业FastAPI项目规范，代码可直接复制进生产模板，半天完成工程化落地。

一、阶段学习目标

1. 使用pydantic-settings分离多环境数据库配置，密码敏感字段加密存储；
2. 全局单例Engine、Session依赖封装，统一连接池参数；
3. 通用BaseCRUD父类封装，所有业务模型复用增删改查/分页；
4. 标准SQLModel项目分层目录（config/database/models/crud/schemas）；
5. Alembic完整初始化、适配SQLModel元数据、生成迁移脚本；
6. 迁移升级/降级、新增字段/删除字段/修改字段实战；
7. 生产环境数据库上线规范，禁止create_all的替代方案。

二、第一部分：工程化分层封装（企业标准项目结构）

2.1 标准项目目录

sqlmodel-demo/ ├── .env # 开发环境配置 ├── .env.prod # 生产环境配置 ├── alembic/ # 数据库迁移目录 ├── alembic.ini # 迁移配置 ├── config/ │ └── settings.py # pydantic-settings全局配置 ├── database/ │ └── session.py # engine、会话生成器 ├── models/ # 数据库实体(table=True) │ ├── user.py │ └── order.py ├── schemas/ # DTO分层模型（Create/Update/Public） │ ├── user_schema.py │ └── order_schema.py ├── crud/ # 业务CRUD，继承通用BaseCRUD │ ├── base.py # 通用父类 │ ├── user_crud.py │ └── order_crud.py └── main.py # 入口测试

2.2 步骤1：多环境配置 pydantic-settings

安装依赖

pipinstallsqlmodel pydantic-settings python-dotenv

2.2.1 .env 开发配置文件

# .env APP_ENV=dev DEBUG=True # 数据库配置 DB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=root DB_PASSWORD=123456 DB_NAME=sql_demo # sqlite可写 DB_URL=sqlite:///./dev.db

2.2.2 config/settings.py 配置模型

importosfrompydanticimportSecretStr,PostgresDsn,MySQLDsnfrompydantic_settingsimportBaseSettings,SettingsConfigDictclassDBSettings(BaseSettings):host:strport:intuser:strpassword:SecretStr# 敏感密码隐藏打印db_name:strmodel_config=SettingsConfigDict(env_prefix="DB_")@propertydefmysql_url(self)->MySQLDsn:"""拼接完整mysql连接字符串"""returnf"mysql+pymysql://{self.user}:{self.password.get_secret_value()}@{self.host}:{self.port}/{self.db_name}?charset=utf8mb4"classGlobalSettings(BaseSettings):model_config=SettingsConfigDict(env_file=".env",env_file_encoding="utf-8",extra="ignore")app_env:strdebug:booldb:DBSettings=DBSettings()# 全局单例配置settings=GlobalSettings()

特点：

• SecretStr隐藏密码，打印不会泄露明文；
• 自动读取.env，环境变量可覆盖配置；
• 拆分DB子配置，结构清晰。

2.3 步骤2：全局Engine & Session封装 database/session.py

fromsqlmodelimportcreate_engine,Sessionfromconfig.settingsimportsettings# 根据环境区分连接参数ifsettings.app_env=="dev":engine=create_engine(settings.db.mysql_url,echo=True,# 开发打印SQLpool_size=5,max_overflow=10)else:engine=create_engine(settings.db.mysql_url,echo=False,pool_size=20,max_overflow=30)# 获取会话生成器（FastAPI依赖注入标准写法）defget_db():withSession(engine)assession:yieldsession

2.4 步骤3：通用BaseCRUD父类 crud/base.py

所有业务CRUD继承，不用重复写新增、分页、查询、删除逻辑

fromtypingimportType,TypeVar,Optional,List,GenericfromsqlmodelimportSQLModel,Session,select,func,update,delete ModelType=TypeVar("ModelType",bound=SQLModel)CreateSchemaType=TypeVar("CreateSchemaType",bound=SQLModel)classBaseCRUD(Generic[ModelType,CreateSchemaType]):def__init__(self,model:Type[ModelType]):self.model=model# 根据主键查询defget(self,db:Session,id:int)->Optional[ModelType]:returndb.get(self.model,id)# 分页查询defget_page(self,db:Session,page:int=1,page_size:int=10):offset=(page-1)*page_size stmt=select(self.model).offset(offset).limit(page_size)items=db.exec(stmt).all()total=db.exec(select(func.count(self.model.id))).scalar()return{"items":items,"total":total,"page":page,"page_size":page_size}# 新增数据defcreate(self,db:Session,obj_in:CreateSchemaType)->ModelType:db_obj=self.model.model_validate(obj_in)db.add(db_obj)db.commit()db.refresh(db_obj)returndb_obj# 局部更新（字典传入更新字段）defupdate(self,db:Session,db_obj:Model,update_data:dict):fork,vinupdate_data.items():ifhasattr(db_obj,k):setattr(db_obj,k,v)db.commit()db.refresh(db_obj)returndb_obj# 删除defremove(self,db:Session,id:int):obj=self.get(db,id)ifobj:db.delete(obj)db.commit()returnobj

2.5 业务CRUD示例 crud/user_crud.py

fromcrud.baseimportBaseCRUDfrommodels.userimportUserfromschemas.user_schemaimportUserCreate# 直接继承通用CRUD，扩展自定义方法即可classUserCRUD(BaseCRUD[User,UserCreate]):defget_by_username(self,db:Session,username:str):stmt=select(User).where(User.username==username)returndb.exec(stmt).first()user_crud=UserCRUD(User)

2.6 模型与DTO分层示例

models/user.py（数据库实体）

fromsqlmodelimportSQLModel,FieldfromtypingimportOptionalfromdatetimeimportdatetimeclassUser(SQLModel,table=True):id:Optional[int]=Field(default=None,primary_key=True)username:str=Field(min_length=3,unique=True)email:strpassword:str=Field(exclude=True)create_time:datetime=Field(default_factory=datetime.utcnow)

schemas/user_schema.py（DTO）

fromsqlmodelimportSQLModelfrompydanticimportEmailStrclassUserCreate(SQLModel):username:stremail:EmailStr password:strclassUserPublic(SQLModel):id:intusername:stremail:str

三、第二部分：Alembic数据库迁移（生产唯一改表方案）

3.1 为什么不能用 create_all()

1. create_all只能新建不存在的表，新增字段/修改字段/删除字段不会同步；
2. 线上多人协作无版本记录，无法回滚结构变更；
3. 生产环境直接运行会覆盖风险，必须版本化迁移工具Alembic。

3.2 安装 & 初始化Alembic

pipinstallalembic# 初始化迁移目录alembic init alembic

生成文件：alembic/文件夹、alembic.ini配置文件

3.3 关键配置修改（适配SQLModel）

3.3.1 alembic.ini

修改文件命名格式，方便区分版本：

[alembic] script_location = alembic file_template = %%(year)d%%(month).2d_%%(slug)s_%%(rev)s # 数据库url交给env.py读取，此处注释 # sqlalchemy.url = xxx

3.3.2 alembic/env.py（核心配置）

修改三处：导入SQLModel、读取项目配置、绑定元数据target_metadata

fromlogging.configimportfileConfigfromsqlalchemyimportengine_from_config,poolfromalembicimportcontext# 导入项目配置与SQLModelfromconfig.settingsimportsettingsfromsqlmodelimportSQLModel# 【必须导入所有models，否则迁移识别不到表】frommodels.userimportUserfrommodels.orderimportOrder config=context.configifconfig.config_file_nameisnotNone:fileConfig(config.config_file_name)# 从配置读取数据库url，不写死db_url=settings.db.mysql_url config.set_main_option("sqlalchemy.url",db_url)# 绑定SQLModel元数据target_metadata=SQLModel.metadata# 下面自动生成的run_migrations_offline/online函数无需修改

3.4 迁移完整命令流程

3.4.1 生成初始迁移（第一次建表）

alembic revision--autogenerate-m"init all tables"

• --autogenerate自动对比模型与数据库差异生成脚本；
• -m填写版本备注，方便维护。

3.4.2 执行升级（应用变更到数据库）

alembic upgradehead

head代表最新版本。

3.4.3 新增字段/修改表后，再次生成迁移

alembic revision--autogenerate-m"add user phone column"alembic upgradehead

3.4.4 版本回滚（线上出错降级）

# 回退1个版本alembic downgrade-1# 指定版本号回退alembic downgrade xxxxxx

3.4.5 迁移注意事项

1. 每次改模型必须执行autogenerate生成脚本，提交代码仓库；
2. 自动生成脚本后务必打开检查，复杂字段（枚举、索引）自动识别可能出错；
3. 生产执行upgrade前先备份数据库；
4. 多对多中间表、联合索引需要手动校验迁移脚本op.create_index逻辑。

四、完整测试入口 main.py

fromsqlmodelimportSessionfromdatabase.sessionimportget_dbfromcrud.user_crudimportuser_crudfromschemas.user_schemaimportUserCreate# 获取数据库会话db=next(get_db())# 新增用户create_data=UserCreate(username="testuser",email="test@qq.com",password="Abc123456")new_user=user_crud.create(db,create_data)print("新增用户ID：",new_user.id)# 分页查询page_data=user_crud.get_page(db,page=1,page_size=10)print("分页数据：",page)

五、阶段核心总结（生产必背规范）

1. 配置分层：使用pydantic-settings拆分多环境，密码用SecretStr脱敏，禁止硬编码数据库地址；
2. 会话统一：全局单例engine，开发开启echo，生产调大连接池；
3. 通用CRUD：BaseCRUD封装分页/新增/查询/删除，业务仅写自定义查询；
4. 项目分层：config/database/models/schemas/crud五层分离，符合SOLID；
5. 迁移规范：线上禁用create_all，统一Alembic版本管理；
6. 迁移流程：修改模型→autogenerate生成脚本→upgrade上线，出错downgrade回滚。

六、生产避坑指南

1. ❌ 数据库密码明文写在代码/ini，使用.env+SecretStr保护；
2. ❌ 每个文件重复创建Session，统一依赖注入get_db；
3. ❌ 线上使用create_all同步表结构，丢失字段无回滚；
4. ❌ 生成迁移脚本不检查，自动识别索引/枚举容易缺失；
5. ❌ 开发、生产共用一套数据库连接参数，未做环境隔离；
6. ✅ 所有业务CRUD继承BaseCRUD，减少80%重复代码；
7. ✅ 项目提交代码时同步提交alembic版本脚本。