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

这里写目录标题前言一、阶段学习目标二、第一部分工程化分层封装企业标准项目结构2.1 标准项目目录2.2 步骤1多环境配置 pydantic-settings2.2.1 .env 开发配置文件2.2.2 config/settings.py 配置模型2.3 步骤2全局Engine Session封装 database/session.py2.4 步骤3通用BaseCRUD父类 crud/base.py2.5 业务CRUD示例 crud/user_crud.py2.6 模型与DTO分层示例三、第二部分Alembic数据库迁移生产唯一改表方案3.1 为什么不能用 create_all()3.2 安装 初始化Alembic3.3 关键配置修改适配SQLModel3.3.1 alembic.ini3.3.2 alembic/env.py核心配置3.4 迁移完整命令流程3.4.1 生成初始迁移第一次建表3.4.2 执行升级应用变更到数据库3.4.3 新增字段/修改表后再次生成迁移3.4.4 版本回滚线上出错降级3.4.5 迁移注意事项四、完整测试入口 main.py五、阶段核心总结生产必背规范六、生产避坑指南前言前面四篇我们掌握了单表、关联、高级查询、事务等零散数据库语法但代码直接堆在一个文件里上线维护会灾难数据库引擎、Session 到处重复定义无法统一管理连接池每个模型手写重复CRUD新增表就要复制一套增删改代码配置硬编码在代码里开发/测试/生产环境切换繁琐线上不能使用create_all()自动建表表结构变更无版本记录无法回滚。本阶段解决全部生产痛点分为两大核心模块项目工程分层封装统一配置、全局会话、通用CRUD父类、标准目录结构Alembic数据库版本迁移SQLModel配套官方迁移工具线上唯一标准改表方案。全程贴合企业FastAPI项目规范代码可直接复制进生产模板半天完成工程化落地。一、阶段学习目标使用pydantic-settings分离多环境数据库配置密码敏感字段加密存储全局单例Engine、Session依赖封装统一连接池参数通用BaseCRUD父类封装所有业务模型复用增删改查/分页标准SQLModel项目分层目录config/database/models/crud/schemasAlembic完整初始化、适配SQLModel元数据、生成迁移脚本迁移升级/降级、新增字段/删除字段/修改字段实战生产环境数据库上线规范禁止create_all的替代方案。二、第一部分工程化分层封装企业标准项目结构2.1 标准项目目录sqlmodel-demo/ ├── .env # 开发环境配置 ├── .env.prod # 生产环境配置 ├── alembic/ # 数据库迁移目录 ├── alembic.ini # 迁移配置 ├── config/ │ └── settings.py # pydantic-settings全局配置 ├── database/ │ └── session.py # engine、会话生成器 ├── models/ # 数据库实体(tableTrue) │ ├── 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-dotenv2.2.1 .env 开发配置文件# .env APP_ENVdev DEBUGTrue # 数据库配置 DB_HOST127.0.0.1 DB_PORT3306 DB_USERroot DB_PASSWORD123456 DB_NAMEsql_demo # sqlite可写 DB_URLsqlite:///./dev.db2.2.2 config/settings.py 配置模型importosfrompydanticimportSecretStr,PostgresDsn,MySQLDsnfrompydantic_settingsimportBaseSettings,SettingsConfigDictclassDBSettings(BaseSettings):host:strport:intuser:strpassword:SecretStr# 敏感密码隐藏打印db_name:strmodel_configSettingsConfigDict(env_prefixDB_)propertydefmysql_url(self)-MySQLDsn:拼接完整mysql连接字符串returnfmysqlpymysql://{self.user}:{self.password.get_secret_value()}{self.host}:{self.port}/{self.db_name}?charsetutf8mb4classGlobalSettings(BaseSettings):model_configSettingsConfigDict(env_file.env,env_file_encodingutf-8,extraignore)app_env:strdebug:booldb:DBSettingsDBSettings()# 全局单例配置settingsGlobalSettings()特点SecretStr隐藏密码打印不会泄露明文自动读取.env环境变量可覆盖配置拆分DB子配置结构清晰。2.3 步骤2全局Engine Session封装 database/session.pyfromsqlmodelimportcreate_engine,Sessionfromconfig.settingsimportsettings# 根据环境区分连接参数ifsettings.app_envdev:enginecreate_engine(settings.db.mysql_url,echoTrue,# 开发打印SQLpool_size5,max_overflow10)else:enginecreate_engine(settings.db.mysql_url,echoFalse,pool_size20,max_overflow30)# 获取会话生成器FastAPI依赖注入标准写法defget_db():withSession(engine)assession:yieldsession2.4 步骤3通用BaseCRUD父类 crud/base.py所有业务CRUD继承不用重复写新增、分页、查询、删除逻辑fromtypingimportType,TypeVar,Optional,List,GenericfromsqlmodelimportSQLModel,Session,select,func,update,delete ModelTypeTypeVar(ModelType,boundSQLModel)CreateSchemaTypeTypeVar(CreateSchemaType,boundSQLModel)classBaseCRUD(Generic[ModelType,CreateSchemaType]):def__init__(self,model:Type[ModelType]):self.modelmodel# 根据主键查询defget(self,db:Session,id:int)-Optional[ModelType]:returndb.get(self.model,id)# 分页查询defget_page(self,db:Session,page:int1,page_size:int10):offset(page-1)*page_size stmtselect(self.model).offset(offset).limit(page_size)itemsdb.exec(stmt).all()totaldb.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_objself.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):objself.get(db,id)ifobj:db.delete(obj)db.commit()returnobj2.5 业务CRUD示例 crud/user_crud.pyfromcrud.baseimportBaseCRUDfrommodels.userimportUserfromschemas.user_schemaimportUserCreate# 直接继承通用CRUD扩展自定义方法即可classUserCRUD(BaseCRUD[User,UserCreate]):defget_by_username(self,db:Session,username:str):stmtselect(User).where(User.usernameusername)returndb.exec(stmt).first()user_crudUserCRUD(User)2.6 模型与DTO分层示例models/user.py数据库实体fromsqlmodelimportSQLModel,FieldfromtypingimportOptionalfromdatetimeimportdatetimeclassUser(SQLModel,tableTrue):id:Optional[int]Field(defaultNone,primary_keyTrue)username:strField(min_length3,uniqueTrue)email:strpassword:strField(excludeTrue)create_time:datetimeField(default_factorydatetime.utcnow)schemas/user_schema.pyDTOfromsqlmodelimportSQLModelfrompydanticimportEmailStrclassUserCreate(SQLModel):username:stremail:EmailStr password:strclassUserPublic(SQLModel):id:intusername:stremail:str三、第二部分Alembic数据库迁移生产唯一改表方案3.1 为什么不能用 create_all()create_all只能新建不存在的表新增字段/修改字段/删除字段不会同步线上多人协作无版本记录无法回滚结构变更生产环境直接运行会覆盖风险必须版本化迁移工具Alembic。3.2 安装 初始化Alembicpipinstallalembic# 初始化迁移目录alembic init alembic生成文件alembic/文件夹、alembic.ini配置文件3.3 关键配置修改适配SQLModel3.3.1 alembic.ini修改文件命名格式方便区分版本[alembic] script_location alembic file_template %%(year)d%%(month).2d_%%(slug)s_%%(rev)s # 数据库url交给env.py读取此处注释 # sqlalchemy.url xxx3.3.2 alembic/env.py核心配置修改三处导入SQLModel、读取项目配置、绑定元数据target_metadatafromlogging.configimportfileConfigfromsqlalchemyimportengine_from_config,poolfromalembicimportcontext# 导入项目配置与SQLModelfromconfig.settingsimportsettingsfromsqlmodelimportSQLModel# 【必须导入所有models否则迁移识别不到表】frommodels.userimportUserfrommodels.orderimportOrder configcontext.configifconfig.config_file_nameisnotNone:fileConfig(config.config_file_name)# 从配置读取数据库url不写死db_urlsettings.db.mysql_url config.set_main_option(sqlalchemy.url,db_url)# 绑定SQLModel元数据target_metadataSQLModel.metadata# 下面自动生成的run_migrations_offline/online函数无需修改3.4 迁移完整命令流程3.4.1 生成初始迁移第一次建表alembic revision--autogenerate-minit all tables--autogenerate自动对比模型与数据库差异生成脚本-m填写版本备注方便维护。3.4.2 执行升级应用变更到数据库alembic upgradeheadhead代表最新版本。3.4.3 新增字段/修改表后再次生成迁移alembic revision--autogenerate-madd user phone columnalembic upgradehead3.4.4 版本回滚线上出错降级# 回退1个版本alembic downgrade-1# 指定版本号回退alembic downgrade xxxxxx3.4.5 迁移注意事项每次改模型必须执行autogenerate生成脚本提交代码仓库自动生成脚本后务必打开检查复杂字段枚举、索引自动识别可能出错生产执行upgrade前先备份数据库多对多中间表、联合索引需要手动校验迁移脚本op.create_index逻辑。四、完整测试入口 main.pyfromsqlmodelimportSessionfromdatabase.sessionimportget_dbfromcrud.user_crudimportuser_crudfromschemas.user_schemaimportUserCreate# 获取数据库会话dbnext(get_db())# 新增用户create_dataUserCreate(usernametestuser,emailtestqq.com,passwordAbc123456)new_useruser_crud.create(db,create_data)print(新增用户ID,new_user.id)# 分页查询page_datauser_crud.get_page(db,page1,page_size10)print(分页数据,page)五、阶段核心总结生产必背规范配置分层使用pydantic-settings拆分多环境密码用SecretStr脱敏禁止硬编码数据库地址会话统一全局单例engine开发开启echo生产调大连接池通用CRUDBaseCRUD封装分页/新增/查询/删除业务仅写自定义查询项目分层config/database/models/schemas/crud五层分离符合SOLID迁移规范线上禁用create_all统一Alembic版本管理迁移流程修改模型→autogenerate生成脚本→upgrade上线出错downgrade回滚。六、生产避坑指南❌ 数据库密码明文写在代码/ini使用.envSecretStr保护❌ 每个文件重复创建Session统一依赖注入get_db❌ 线上使用create_all同步表结构丢失字段无回滚❌ 生成迁移脚本不检查自动识别索引/枚举容易缺失❌ 开发、生产共用一套数据库连接参数未做环境隔离✅ 所有业务CRUD继承BaseCRUD减少80%重复代码✅ 项目提交代码时同步提交alembic版本脚本。