一、项目背景“开发环境跑不起来”——这是星云电商订单中台团队过去三个月最常见的群消息。上一季度新入职的小王花了整整三天才把项目跑通。原因包括PostgreSQL 版本不匹配本地 14生产 16、Python 依赖冲突有人装了 SQLAlchemy 1.4有人装了 2.0、Alembic 迁移脚本在不同机器上产生不同的 revision 编号。更糟糕的是因为没有统一的 Docker 环境CI 流水线上跑的测试和本地开发环境的行为不一致——测试在 CI 上挂了一片但在小李的 MacBook 上全绿。另一个隐蔽的坑是日志。开发期间为了排查问题团队在代码里到处加了print(conn_info)和echoTrue。结果有一次代码带着echoTrue上了生产数据库每执行一条 SQL 都往标准输出打日志不仅是性能灾难日志 IO 比 SQL 本身还慢还把用户的手机号、密码 hash 全打进了 log 文件里。第三个痛点是项目目录结构混乱。models.py一个文件 3000 行里面塞了 20 个模型的类定义、用到的工具函数、甚至还有几个测试用例。运维在做迁移时根本找不到哪个文件是迁移入口测试同学写的 fixture 和业务代码紧紧耦合在一起。大师意识到在开始真正写 SQLAlchemy 代码之前必须先解决环境一致性和工程结构约定这两个前置问题。本章的目标就是搭建一个可复现、可移植的开发环境并为后续 14 章的代码建立统一的目录骨架。二、项目设计场景周二上午工区茶水间。小胖刚从食堂吃完早饭回来手里还端着杯豆浆。小白已经坐在工位上屏幕上是密密麻麻的 Docker 文档。小胖“小白你一大早就看 Docker 文档啊我以为我们就写个 Python 脚本连数据库就行怎么还要 Docker这不是把简单问题搞复杂了吗——就像吃个泡面还要先装修厨房”小白头也不抬“那你有没有试过换台电脑跑你的脚本比如你本地 PostgreSQL 16我本地 PostgreSQL 14你写的 SQL 里用了 PostgreSQL 16 才支持的语法到我这里就挂了。更何况生产环境 Ubuntu PostgreSQL 16测试 CI 是 Alpine PostgreSQL 15……”小胖“呃……我确实在测试环境碰到过SELECT ... FOR UPDATE SKIP LOCKED在老版本上不支持当时查了半天。”大师端着咖啡走过来“这就是我们要用 Docker Compose 的原因。Docker 不是装修厨房是打包一个人人都一样的套餐。你想想如果食堂外包给不同厨师每个厨师炒菜的方式不一样你是不是每次吃饭都有惊喜Docker 就是把厨师也标准化了。”小胖“技术映射Docker Compose 标准化环境打包工具。行吧那 Docker Compose 文件里要写哪些东西”大师“至少两个服务一个 PostgreSQL 数据库容器如果需要的话还有一个 Python 应用容器。数据库容器定义了 PostgreSQL 的版本和初始配置应用容器里跑我们的代码。这样所有人——开发、测试、运维——拉下来就是一模一样的环境。”小白“那 python 依赖怎么管我之前见过有人用pip freeze requirements.txt结果导出了 200 个包根本不知道哪些是真正需要的哪些是间接依赖。”大师“用pyproject.toml或setup.cfg把直接依赖明确列出来版本范围也要精确。像 SQLAlchemy 我们锁2.0.30,2.1Alembic 锁1.13,1.14。不要用pip freeze裸导出那是懒人做法但也是隐患源头。然后配上虚拟环境.venv/进.gitignore。”小胖“那目录结构呢我们现在就一个models.py什么都在里面……”大师“这是我要重点讲的。一个可维护的 SQLAlchemy 项目的目录结构应该像这样——”大师打开笔记本画了一个目录树nebula-order-center/ ├── pyproject.toml # 项目元数据 依赖声明 ├── docker-compose.yml # 本地开发数据库环境 ├── Dockerfile # 可选应用容器化 ├── .env.example # 环境变量模板 ├── .gitignore ├── src/ │ └── order_center/ │ ├── __init__.py │ ├── config.py # 全局配置DB URL、pool 参数等 │ ├── db/ │ │ ├── __init__.py │ │ ├── engine.py # create_engine 工厂 │ │ └── session.py # Session 工厂 │ ├── models/ # 每个文件一个聚合根模型 │ │ ├── __init__.py │ │ ├── base.py # DeclarativeBase MetaData │ │ ├── user.py │ │ ├── product.py │ │ └── order.py │ ├── repository/ # 数据访问封装可选 │ ├── service/ # 业务逻辑 │ └── utils/ ├── alembic/ # 迁移脚本目录第14章细讲 │ ├── env.py │ ├── versions/ │ └── alembic.ini ├── tests/ │ ├── conftest.py # pytest fixture引擎、会话等 │ ├── factories/ # 测试数据工厂 │ ├── test_models/ │ └── test_repository/ └── scripts/ # 运维脚本 └── health_check.py小白“我对db/engine.py和db/session.py的分开很感兴趣。也就是说Engine 是全局单例Session 是每次请求新建的”大师“完全正确。Engine 是一个重对象——它内部有连接池、Dialect 实例、配置信息。整个进程只需要一个 Engine 实例。而 Session 是轻量的每个 Web 请求或工作单元创建一个用完即弃。”小胖“就像食堂的厨房是固定的Engine但每个人吃饭的餐盘是临时拿的Session吃完就还回去”大师“技术映射Engine 全局厨房Session 临时餐盘。这个比喻好。记住Engine 的创建成本高但持有成本低Session 的创建成本低但持有期间可能持有连接。”小白“还有一个我一直想问的为什么目录里要分models/user.py和repository/两个层直接把查询写在 Service 里不就行了吗”大师“这个问题很关键。小胖你来说说你把查询直接写在 Service 里遇到过什么问题”小胖“呃……有一次我改了一个下单的查询逻辑忘了还有一个’订单催单’功能也用了类似的 WHERE 条件。结果上线后催单功能挂了因为那个 SQL 被我改错了但没人发现。”大师“这就是分层的好处。Repository 层专门负责数据访问——构造查询、执行查询、返回结果。Service 层只关心业务逻辑不关心数据从哪来怎么查。当查询逻辑需要变更时只改 Repository 层Service 层不受影响。当你想从 PostgreSQL 换成其他数据库时也只需要改 Repository 层。”小白“技术映射Repository 模式 数据访问隔离层。那测试呢Test 目录里的 conftest.py 是干什么的”大师“这是 pytest 的 fixture 控制中心。里面定义引擎、会话、数据库初始化的 fixture每个测试函数需要数据库时直接通过参数注入。最重要的是——我们用事务回滚机制保证每个测试之间数据隔离开一个外层事务测试在里面跑跑完回滚。测试不脏库并行跑互不干扰。”小胖“技术映射Fixture 事务回滚 测试数据隔离沙箱。这和我在本地开个测试库乱跑的区别是啥”大师“速度。事务回滚比 truncate 表快 10 倍以上而且不需要维护测试库的表结构——表从 migration 来测试只是不 commit 而已。”小白“最后一个问题.env.example放什么连接串和密码怎么办”大师“绝不要把密码写进代码里。.env.example存模板不包含真实密码。真实密码从环境变量、Kubernetes Secret、Vault 或 CI Secret 读取。config.py里用os.getenv(DATABASE_URL, ...)读取再传给create_engine。”小胖“明白了。这不是装修厨房是建一个标准化的中央厨房——配方依赖、炉灶Docker、管理流程目录约定全标准化。”三、项目实战实战目标初始化星云订单中台空仓库用 Docker Compose 拉起 PostgreSQL跑通健康检查 SQL验证环境完整可用。环境准备# 确认工具链python--version# 3.11docker--version# 24dockercompose version# 2# 创建项目目录mkdirnebula-order-centercdnebula-order-center步骤一编写 Docker Compose 文件# docker-compose.ymlversion:3.9services:db:image:postgres:16-alpinecontainer_name:nebula-pgenvironment:POSTGRES_USER:nebulaPOSTGRES_PASSWORD:nebula_dev# 仅开发环境生产用环境变量注入POSTGRES_DB:order_centerports:-5432:5432volumes:-pgdata:/var/lib/postgresql/data# 可选初始 SQL 脚本-./scripts/init.sql:/docker-entrypoint-initdb.d/init.sqlhealthcheck:test:[CMD-SHELL,pg_isready -U nebula -d order_center]interval:5stimeout:5sretries:5volumes:pgdata:-- scripts/init.sql可选-- 创建初始 SchemaCREATESCHEMAIFNOTEXISTSorder_center;拉起并验证dockercompose up-d# 后台启动dockercomposeps# 查看容器状态应为 healthydockercompose logs db# 查看 PostgreSQL 日志步骤二配置 Python 项目依赖# pyproject.toml [project] name nebula-order-center version 0.1.0 requires-python 3.11 dependencies [ sqlalchemy[asyncio]2.0.30,2.1, psycopg[binary]3.1,3.3, asyncpg0.29,0.31, alembic1.13,1.15, ] [project.optional-dependencies] dev [ pytest8.0, pytest-asyncio0.23, factory-boy3.3, ipython8.0, ] [tool.pytest.ini_options] testpaths [tests] asyncio_mode auto# .gitignore__pycache__/ *.py[cod].venv/ .env *.egg-info/ dist/ build/ .pytest_cache/ .mypy_cache/# .env.example —— 开发环境变量模板不要提交真实密码DATABASE_URLpostgresqlpsycopg://nebula:nebula_devlocalhost:5432/order_centerDATABASE_URL_ASYNCpostgresqlasyncpg://nebula:nebula_devlocalhost:5432/order_center安装依赖python-mvenv .venvsource.venv/bin/activate# Windows: .venv\Scripts\activatepipinstall-e.[dev]步骤三创建项目骨架# src/order_center/__init__.py星云电商订单中台# src/order_center/config.pyimportos# 数据库配置DATABASE_URLos.getenv(DATABASE_URL,postgresqlpsycopg://nebula:nebula_devlocalhost:5432/order_center)DATABASE_URL_ASYNCos.getenv(DATABASE_URL_ASYNC,postgresqlasyncpg://nebula:nebula_devlocalhost:5432/order_center)# 连接池配置第23章详解POOL_SIZEint(os.getenv(POOL_SIZE,5))MAX_OVERFLOWint(os.getenv(MAX_OVERFLOW,10))POOL_TIMEOUTint(os.getenv(POOL_TIMEOUT,30))POOL_RECYCLEint(os.getenv(POOL_RECYCLE,1800))# 30 min# 是否回显 SQL仅开发环境ECHO_SQLos.getenv(ECHO_SQL,false).lower()true# src/order_center/db/__init__.py数据库连接与 Session 管理# src/order_center/db/engine.pyfromsqlalchemyimportcreate_enginefromorder_center.configimportDATABASE_URL,POOL_SIZE,MAX_OVERFLOW,POOL_TIMEOUT,POOL_RECYCLE,ECHO_SQL enginecreate_engine(DATABASE_URL,echoECHO_SQL,pool_sizePOOL_SIZE,max_overflowMAX_OVERFLOW,pool_timeoutPOOL_TIMEOUT,pool_recyclePOOL_RECYCLE,pool_pre_pingTrue,# 2.0 默认执行前检测连接有效性)# src/order_center/db/session.pyfromsqlalchemy.ormimportsessionmaker,Sessionfromorder_center.db.engineimportengine# sessionmaker 是 Session 工厂SyncSessionFactorysessionmaker(bindengine,autocommitFalse,autoflushFalse,# 不自动 flush由开发者显式控制expire_on_commitTrue,)defget_session()-Session:创建新的同步 Session每个工作单元一个returnSyncSessionFactory()# src/order_center/models/__init__.pyORM 模型层# src/order_center/models/base.pyfromsqlalchemy.ormimportDeclarativeBasefromsqlalchemyimportMetaData# 命名约定建议配合 Alembic 自动检测约束命名naming_convention{ix:ix_%(column_0_label)s,uq:uq_%(table_name)s_%(column_0_name)s,ck:ck_%(table_name)s_%(constraint_name)s,fk:fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s,pk:pk_%(table_name)s,}classBase(DeclarativeBase):metadataMetaData(naming_conventionnaming_convention)步骤四健康检查验证# scripts/health_check.py验证数据库连接、Engine 和 ORM 基础功能是否正常fromorder_center.db.engineimportenginefromorder_center.db.sessionimportget_sessionfromorder_center.models.baseimportBasefromsqlalchemyimporttext,select,funcdefcheck_engine():步骤1验证 Engine 能连上数据库print(*50)print(1. Engine 健康检查)try:withengine.connect()asconn:resultconn.execute(text(SELECT 1)).scalar()print(f [OK] Engine 连接成功SELECT 1 {result})# 检查 PostgreSQL 版本versionconn.execute(text(SELECT version())).scalar()print(f [OK] 数据库版本:{version[:50]}...)exceptExceptionase:print(f [FAIL] Engine 连接失败:{e})returnFalsereturnTruedefcheck_session():步骤2验证 Session 能正常创建和执行print(\n2. Session 健康检查)try:sessionget_session()resultsession.execute(text(SELECT COUNT(*) FROM pg_tables)).scalar()print(f [OK] Session 创建成功当前数据库表数:{result})session.close()exceptExceptionase:print(f [FAIL] Session 检查失败:{e})returnFalsereturnTruedefcheck_orm():步骤3验证 ORM 声明基类可用print(\n3. ORM 基类检查)try:# 不实际建表只验证 Base 类可正常使用tablesBase.metadata.tablesprint(f [OK] Base 基类就绪已注册表:{list(tables.keys())iftableselse(尚无模型)})exceptExceptionase:print(f [FAIL] ORM 基类检查失败:{e})returnFalsereturnTrueif__name____main__:print(*50)print(星云订单中台 —— 环境健康检查)print(*50)results[check_engine(),check_session(),check_orm()]print(\n*50)ifall(results):print(全部检查通过环境搭建完成。)else:print(存在失败项请检查日志排查。)运行健康检查# 先确保 Docker 容器在运行dockercompose up-d# 运行检查脚本python scripts/health_check.py预期输出 星云订单中台 —— 环境健康检查 1. Engine 健康检查 [OK] Engine 连接成功SELECT 1 1 [OK] 数据库版本: PostgreSQL 16.x on x86_64-pc-linux-musl... 2. Session 健康检查 [OK] Session 创建成功当前数据库表数: 73 3. ORM 基类检查 [OK] Base 基类就绪已注册表: (尚无模型) 全部检查通过环境搭建完成。可能遇到的坑及解决方法Docker 端口冲突现象Error starting userland proxy: Ports are not available: listen tcp 0.0.0.0:5432: bind: address already in use原因本机已有 PostgreSQL 占用 5432 端口。解决修改docker-compose.yml中的端口映射为5433:5432并同步修改.env中的连接串端口。psycopg 安装失败现象error: Microsoft Visual C 14.0 is required原因psycopg非 binary 版需要 C 编译环境。解决使用psycopg[binary]版本预编译 wheel已在pyproject.toml中配置。echoTrue无限刷屏现象控制台输出被 SQL 日志淹没无法看到正常输出。解决health_check.py中的 engine 单独创建时不设 echo或通过配置开关控制。.env文件未加载现象环境变量DATABASE_URL未生效使用了默认值。解决使用python-dotenv或 IDE 自带插件加载.env。生产中通过 systemd/kubernetes 注入环境变量。测试验证# tests/conftest.pypytest 全局 fixture —— 提供测试引擎和事务隔离的会话importpytestfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker,Sessionfromorder_center.models.baseimportBase TEST_DB_URLpostgresqlpsycopg://nebula:nebula_devlocalhost:5432/order_centerpytest.fixture(scopesession)defengine():测试引擎整个测试会话复用test_enginecreate_engine(TEST_DB_URL,echoFalse)Base.metadata.create_all(test_engine)# 确保表存在returntest_enginepytest.fixturedefsession(engine):测试会话每个测试函数独立事务回滚保证隔离connectionengine.connect()transactionconnection.begin()# 开启外层事务SessionFactorysessionmaker(bindconnection)sessionSessionFactory()yieldsession session.close()transaction.rollback()# 回滚不污染数据库connection.close()# tests/test_health.py环境搭建验证测试deftest_can_connect_to_database(session):验证测试 session 能正常连接resultsession.execute(session.bind.bind.dbapi_connection.execute# 这行仅示意)# 实际验证fromsqlalchemyimporttext rowsession.execute(text(SELECT 2 2 AS answer)).fetchone()assertrow[0]4deftest_session_is_transactional(session):验证 session 的事务隔离特性fromsqlalchemyimporttext session.execute(text(CREATE TEMP TABLE _test_t (id INT)))session.execute(text(INSERT INTO _test_t VALUES (1)))# 在事务内可见resultsession.execute(text(SELECT COUNT(*) FROM _test_t)).scalar()assertresult1deftest_sessions_are_isolated(engine):验证两个独立的 session 之间数据隔离fromsqlalchemy.ormimportSessionasOrmSession s1OrmSession(engine)s2OrmSession(engine)# ... 隔离性验证逻辑s1.close()s2.close()运行测试pytest tests/-v四、项目总结优点与缺点对比维度无工程约定本文标准化骨架环境一致性每人本地不同CI 可能挂Docker Compose 统一跨平台一致依赖管理pip freeze 裸导出版本冲突pyproject.toml 精确版本 extras目录可发现性一个 models.py 3000 行按域拆分新人 5 分钟定位安全性密码硬编码echo 上生产环境变量注入echo 可配置关闭测试隔离测试互串数据污染事务回滚隔离可并行学习成本低但后续维护成本爆炸初期有学习开销但后续平滑适用场景推荐使用此骨架的场景3 人以上团队协作的 SQLAlchemy 项目。需要跨多环境本地/CI/预发/生产部署的项目。预计生命周期超过一年、需要持续迭代的业务项目。需要测试覆盖率要求的项目。有多数据库环境需求通过 DuckDB/sqlite 做本地轻量测试。不推荐使用的场景个人学习/实验项目——过于繁琐直接单文件跑即可。一次性数据迁移脚本——不需要完整的工程骨架。注意事项pyproject.toml依赖版本要精确不要写2.0而应写2.0.30,2.1防止 CI 自动拉取 major 版本导致不兼容。Docker 的 PostgreSQL 数据是临时的如果删了容器数据也会丢失除非挂载 volume。开发环境的核心数据要持久化或者有迁移脚本。配置文件不要提交到仓库.env必须进.gitignore只提交.env.example模板。Engine 是整个进程的单例不要在函数里每次都create_engine()否则会创建多个连接池导致连接数失控。常见踩坑经验案例 1Docker 时间与宿主机不同步现象数据库中的created_at字段比实际时间慢了 8 小时。根因Docker 容器默认使用 UTC 时区与宿主机时区不同。修复在docker-compose.yml中添加TZ: Asia/Shanghai到 environment 中。案例 2.gitignore遗漏 Alembic 版本目录现象合并代码时 Alembicversions/目录冲突。根因Alembic 生成的迁移脚本必须纳入版本控制属于项目源代码如果.gitignore误过滤了versions/*.py则迁移丢失。修复.gitignore中只忽略__pycache__不要忽略alembic/versions/。案例 3psycopg3 与 psycopg2 混装现象项目中同时import psycopg2和import psycopg连接串混写。根因团队里有人沿用了老的 psycopg2 写法。SQLAlchemy 2.0 推荐 psycopg 3.xpsycopg包名不是psycopg2URL 前缀为postgresqlpsycopg://。修复统一为 psycopg 3.x asyncpg不再引入 psycopg2。思考题某项目在生产环境中每次调用create_engine()都创建了一个引擎实例。这个做法会导致什么问题为什么 Engine 应该被定义为模块级别的单例结合连接池的工作机制解释。团队里一位同事建议把所有Dockerfile、docker-compose.yml、.env、pyproject.toml都放到一个deploy/目录下。这种组织方式有什么利弊你建议在什么场景下采用这种方案延伸阅读与资源NumPy 从入门到生产落地全链路实战指南科学计算/向量化Redis 8 实战精讲从 CRUD 到源码构建高可用缓存系统Redis 实战修炼与原理进阶Python 3实战精进从脚本到高并发订单引擎python入门Rquests从菜鸟脚本到企业级SDK的网络实战圣经Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地MongoDB 实战进阶与内核修炼后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析参考答案参见附录 E。