一、项目背景

“开发环境跑不起来”——这是星云电商订单中台团队过去三个月最常见的群消息。

上一季度，新入职的小王花了整整三天才把项目跑通。原因包括：PostgreSQL 版本不匹配（本地 14，生产 16）、Python 依赖冲突（有人装了 SQLAlchemy 1.4，有人装了 2.0）、Alembic 迁移脚本在不同机器上产生不同的 revision 编号。更糟糕的是，因为没有统一的 Docker 环境，CI 流水线上跑的测试和本地开发环境的行为不一致——测试在 CI 上挂了一片，但在小李的 MacBook 上全绿。

另一个隐蔽的坑是日志。开发期间为了排查问题，团队在代码里到处加了 print(conn_info) 和 echo=True。结果有一次代码带着 echo=True 上了生产，数据库每执行一条 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.1，Alembic 锁 >=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.11
docker --version    # >= 24
docker compose version  # >= 2

# 创建项目目录
mkdir nebula-order-center
cd nebula-order-center

步骤一：编写 Docker Compose 文件

# docker-compose.yml
version: "3.9"
services:
  db:
    image: postgres:16-alpine
    container_name: nebula-pg
    environment:
      POSTGRES_USER: nebula
      POSTGRES_PASSWORD: nebula_dev  # 仅开发环境，生产用环境变量注入
      POSTGRES_DB: order_center
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
      # 可选：初始 SQL 脚本
      - ./scripts/init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U nebula -d order_center"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  pgdata:

-- scripts/init.sql（可选）
-- 创建初始 Schema
CREATE SCHEMA IF NOT EXISTS order_center;

拉起并验证：

docker compose up -d     # 后台启动
docker compose ps        # 查看容器状态：应为 healthy
docker compose 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",
    "asyncpg>=0.29,<0.31",
    "alembic>=1.13,<1.15",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "factory-boy>=3.3",
    "ipython>=8.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_URL=postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center
DATABASE_URL_ASYNC=postgresql+asyncpg://nebula:nebula_dev@localhost:5432/order_center

安装依赖：

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -e ".[dev]"

步骤三：创建项目骨架

# src/order_center/__init__.py
"""星云电商订单中台"""

# src/order_center/config.py
import os

# 数据库配置
DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center"
)
DATABASE_URL_ASYNC = os.getenv(
    "DATABASE_URL_ASYNC",
    "postgresql+asyncpg://nebula:nebula_dev@localhost:5432/order_center"
)

# 连接池配置（第23章详解）
POOL_SIZE = int(os.getenv("POOL_SIZE", "5"))
MAX_OVERFLOW = int(os.getenv("MAX_OVERFLOW", "10"))
POOL_TIMEOUT = int(os.getenv("POOL_TIMEOUT", "30"))
POOL_RECYCLE = int(os.getenv("POOL_RECYCLE", "1800"))  # 30 min

# 是否回显 SQL（仅开发环境）
ECHO_SQL = os.getenv("ECHO_SQL", "false").lower() == "true"

# src/order_center/db/__init__.py
"""数据库连接与 Session 管理"""

# src/order_center/db/engine.py
from sqlalchemy import create_engine
from order_center.config import DATABASE_URL, POOL_SIZE, MAX_OVERFLOW, POOL_TIMEOUT, POOL_RECYCLE, ECHO_SQL

engine = create_engine(
    DATABASE_URL,
    echo=ECHO_SQL,
    pool_size=POOL_SIZE,
    max_overflow=MAX_OVERFLOW,
    pool_timeout=POOL_TIMEOUT,
    pool_recycle=POOL_RECYCLE,
    pool_pre_ping=True,  # 2.0 默认，执行前检测连接有效性
)

# src/order_center/db/session.py
from sqlalchemy.orm import sessionmaker, Session
from order_center.db.engine import engine

# sessionmaker 是 Session 工厂
SyncSessionFactory = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,   # 不自动 flush，由开发者显式控制
    expire_on_commit=True,
)

def get_session() -> Session:
    """创建新的同步 Session（每个工作单元一个）"""
    return SyncSessionFactory()

# src/order_center/models/__init__.py
"""ORM 模型层"""

# src/order_center/models/base.py
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import MetaData

# 命名约定建议（配合 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",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=naming_convention)

步骤四：健康检查验证

# scripts/health_check.py
"""验证数据库连接、Engine 和 ORM 基础功能是否正常"""

from order_center.db.engine import engine
from order_center.db.session import get_session
from order_center.models.base import Base
from sqlalchemy import text, select, func

def check_engine():
    """步骤1：验证 Engine 能连上数据库"""
    print("=" * 50)
    print("1. Engine 健康检查")
    try:
        with engine.connect() as conn:
            result = conn.execute(text("SELECT 1")).scalar()
            print(f"   [OK] Engine 连接成功，SELECT 1 = {result}")
            # 检查 PostgreSQL 版本
            version = conn.execute(text("SELECT version()")).scalar()
            print(f"   [OK] 数据库版本: {version[:50]}...")
    except Exception as e:
        print(f"   [FAIL] Engine 连接失败: {e}")
        return False
    return True

def check_session():
    """步骤2：验证 Session 能正常创建和执行"""
    print("\n2. Session 健康检查")
    try:
        session = get_session()
        result = session.execute(text("SELECT COUNT(*) FROM pg_tables")).scalar()
        print(f"   [OK] Session 创建成功，当前数据库表数: {result}")
        session.close()
    except Exception as e:
        print(f"   [FAIL] Session 检查失败: {e}")
        return False
    return True

def check_orm():
    """步骤3：验证 ORM 声明基类可用"""
    print("\n3. ORM 基类检查")
    try:
        # 不实际建表，只验证 Base 类可正常使用
        tables = Base.metadata.tables
        print(f"   [OK] Base 基类就绪，已注册表: {list(tables.keys()) if tables else '(尚无模型)'}")
    except Exception as e:
        print(f"   [FAIL] ORM 基类检查失败: {e}")
        return False
    return True

if __name__ == "__main__":
    print("=" * 50)
    print("星云订单中台 —— 环境健康检查")
    print("=" * 50)
    results = [check_engine(), check_session(), check_orm()]
    print("\n" + "=" * 50)
    if all(results):
        print("全部检查通过！环境搭建完成。")
    else:
        print("存在失败项，请检查日志排查。")

运行健康检查：

# 先确保 Docker 容器在运行
docker compose 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 基类就绪，已注册表: (尚无模型)
==================================================
全部检查通过！环境搭建完成。

可能遇到的坑及解决方法

  1. 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 中的连接串端口。
  1. psycopg 安装失败
  • 现象：error: Microsoft Visual C++ 14.0 is required
  • 原因：psycopg（非 binary 版）需要 C 编译环境。
  • 解决：使用 psycopg[binary] 版本（预编译 wheel），已在 pyproject.toml 中配置。
  1. echo=True 无限刷屏
  • 现象：控制台输出被 SQL 日志淹没，无法看到正常输出。
  • 解决：health_check.py 中的 engine 单独创建时不设 echo，或通过配置开关控制。
  1. .env 文件未加载
  • 现象：环境变量 DATABASE_URL 未生效，使用了默认值。
  • 解决：使用 python-dotenv 或 IDE 自带插件加载 .env。生产中通过 systemd/kubernetes 注入环境变量。

测试验证

# tests/conftest.py
"""pytest 全局 fixture —— 提供测试引擎和事务隔离的会话"""

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from order_center.models.base import Base

TEST_DB_URL = "postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center"

@pytest.fixture(scope="session")
def engine():
    """测试引擎：整个测试会话复用"""
    test_engine = create_engine(TEST_DB_URL, echo=False)
    Base.metadata.create_all(test_engine)  # 确保表存在
    return test_engine

@pytest.fixture
def session(engine):
    """测试会话：每个测试函数独立，事务回滚保证隔离"""
    connection = engine.connect()
    transaction = connection.begin()       # 开启外层事务
    SessionFactory = sessionmaker(bind=connection)
    session = SessionFactory()
    yield session
    session.close()
    transaction.rollback()                 # 回滚，不污染数据库
    connection.close()

# tests/test_health.py
"""环境搭建验证测试"""

def test_can_connect_to_database(session):
    """验证测试 session 能正常连接"""
    result = session.execute(
        session.bind.bind.dbapi_connection.execute  # 这行仅示意
    )
    # 实际验证：
    from sqlalchemy import text
    row = session.execute(text("SELECT 2 + 2 AS answer")).fetchone()
    assert row[0] == 4

def test_session_is_transactional(session):
    """验证 session 的事务隔离特性"""
    from sqlalchemy import text
    session.execute(text("CREATE TEMP TABLE _test_t (id INT)"))
    session.execute(text("INSERT INTO _test_t VALUES (1)"))
    # 在事务内可见
    result = session.execute(text("SELECT COUNT(*) FROM _test_t")).scalar()
    assert result == 1

def test_sessions_are_isolated(engine):
    """验证两个独立的 session 之间数据隔离"""
    from sqlalchemy.orm import Session as OrmSession
    s1 = OrmSession(engine)
    s2 = OrmSession(engine)
    # ... 隔离性验证逻辑
    s1.close()
    s2.close()

运行测试：

pytest tests/ -v

四、项目总结

优点与缺点

对比维度 无工程约定 本文标准化骨架
环境一致性 每人本地不同，CI 可能挂 Docker Compose 统一，跨平台一致
依赖管理 pip freeze 裸导出，版本冲突 pyproject.toml 精确版本 + extras
目录可发现性 一个 models.py 3000 行 按域拆分，新人 5 分钟定位
安全性 密码硬编码，echo 上生产 环境变量注入，echo 可配置关闭
测试隔离 测试互串，数据污染 事务回滚隔离，可并行
学习成本 低，但后续维护成本爆炸 初期有学习开销，但后续平滑

适用场景

推荐使用此骨架的场景：

  1. 3 人以上团队协作的 SQLAlchemy 项目。
  2. 需要跨多环境（本地/CI/预发/生产）部署的项目。
  3. 预计生命周期超过一年、需要持续迭代的业务项目。
  4. 需要测试覆盖率要求的项目。
  5. 有多数据库环境需求（通过 DuckDB/sqlite 做本地轻量测试）。

不推荐使用的场景：

  1. 个人学习/实验项目——过于繁琐，直接单文件跑即可。
  2. 一次性数据迁移脚本——不需要完整的工程骨架。

注意事项

  1. pyproject.toml 依赖版本要精确：不要写 >=2.0，而应写 >=2.0.30,<2.1，防止 CI 自动拉取 major 版本导致不兼容。
  2. Docker 的 PostgreSQL 数据是临时的：如果删了容器，数据也会丢失（除非挂载 volume）。开发环境的核心数据要持久化或者有迁移脚本。
  3. 配置文件不要提交到仓库：.env 必须进 .gitignore，只提交 .env.example 模板。
  4. Engine 是整个进程的单例：不要在函数里每次都 create_engine()，否则会创建多个连接池导致连接数失控。

常见踩坑经验

案例 1：Docker 时间与宿主机不同步

  • 现象：数据库中的 created_at 字段比实际时间慢了 8 小时。
  • 根因：Docker 容器默认使用 UTC 时区，与宿主机时区不同。
  • 修复：在 docker-compose.yml 中添加 TZ: Asia/Shanghai 到 environment 中。

案例 2：.gitignore 遗漏 Alembic 版本目录

  • 现象：合并代码时 Alembic versions/ 目录冲突。
  • 根因：Alembic 生成的迁移脚本必须纳入版本控制（属于项目源代码），如果 .gitignore 误过滤了 versions/*.py，则迁移丢失。
  • 修复：.gitignore 中只忽略 __pycache__，不要忽略 alembic/versions/。

案例 3：psycopg3 与 psycopg2 混装

  • 现象：项目中同时 import psycopg2 和 import psycopg，连接串混写。
  • 根因：团队里有人沿用了老的 psycopg2 写法。SQLAlchemy 2.0 推荐 psycopg 3.x（psycopg 包名，不是 psycopg2），URL 前缀为 postgresql+psycopg://。
  • 修复：统一为 psycopg 3.x + asyncpg，不再引入 psycopg2。

思考题

  1. 某项目在生产环境中，每次调用 create_engine() 都创建了一个引擎实例。这个做法会导致什么问题？为什么 Engine 应该被定义为模块级别的单例？结合连接池的工作机制解释。

  2. 团队里一位同事建议把所有 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。

Logo
快递鸟一站式物流API解决方案

电商企业物流数字化转型必备！快递鸟 API 接口，72 小时快速完成物流系统集成。全流程实战1V1指导，营造开放的API技术生态圈。

更多推荐

cover

微信AI掘金指南：从流量获取到商业变现

avatar 快递鸟社区
cover

深入读懂微信 AI 底层逻辑：普通人、商家掘金完整手册

avatar 快递鸟社区
cover

读懂微信 AI 底层逻辑：普通人、商家掘金完整手册

avatar 快递鸟社区