一套稳健的 FastAPI 工程,不是把 API、Redis 和 PostgreSQL 塞进同一个 Compose 文件就算完事。更合理的做法是把应用设计成无状态服务,数据库迁移、配置管理、健康检查和测试各自承担清晰职责;开发环境追求反馈速度,生产环境则强调镜像可复现、最小权限和可观测性。
下面给出一套可以直接落地的项目骨架。
一、推荐架构
开发环境可以使用 Docker Compose 编排所有依赖,代码通过挂载目录实现热更新。生产环境仍然使用相同镜像,但不再挂载源代码,也不启用 --reload。
1flowchart LR 2 C[客户端] --> N[Nginx / Traefik / LB] 3 N --> A[FastAPI 容器] 4 A --> P[(PostgreSQL)] 5 A --> R[(Redis)] 6 M[Alembic 迁移任务] --> P 7 W[后台任务 Worker] --> P 8 W --> R 9
各组件的职责建议如下。
| 组件 | 主要职责 | 工程建议 |
|---|---|---|
| FastAPI | HTTP 接口、业务编排 | 保持无状态,不在本地磁盘保存业务数据 |
| PostgreSQL | 持久化业务数据 | 使用 SQLAlchemy 2.x、Alembic |
| Redis | 缓存、限流、分布式锁 | 不把 Redis 当作唯一可靠数据源 |
| Worker | 执行耗时任务 | 与 API 使用同一代码库、不同启动命令 |
| Alembic | 管理数据库结构 | 作为独立的一次性任务运行 |
| 反向代理 | TLS、路由、限流 | 云环境也可以交给托管负载均衡器 |
FastAPI 官方现在更推荐从标准 Python 镜像自行构建,而不是依赖过去的 tiangolo/uvicorn-gunicorn-fastapi 基础镜像。容器环境中通常让每个容器运行一个应用进程,再由编排平台完成横向扩容;单机 Docker 部署则可以根据 CPU 和数据库连接预算设置多个 worker。
二、项目目录设计
目录不宜过早拆得过细,但数据库、缓存、配置和业务领域应当分开。
1fastapi-project/ 2├── app/ 3│ ├── __init__.py 4│ ├── main.py 5│ ├── api/ 6│ │ ├── __init__.py 7│ │ ├── deps.py 8│ │ └── v1/ 9│ │ ├── router.py 10│ │ └── endpoints/ 11│ │ └── users.py 12│ ├── core/ 13│ │ ├── config.py 14│ │ ├── logging.py 15│ │ └── security.py 16│ ├── db/ 17│ │ ├── base.py 18│ │ ├── models.py 19│ │ └── session.py 20│ ├── repositories/ 21│ │ └── user.py 22│ ├── schemas/ 23│ │ └── user.py 24│ └── services/ 25│ └── user.py 26├── migrations/ 27├── tests/ 28│ ├── conftest.py 29│ ├── integration/ 30│ └── unit/ 31├── alembic.ini 32├── compose.yaml 33├── compose.dev.yaml 34├── Dockerfile 35├── pyproject.toml 36├── uv.lock 37├── .dockerignore 38├── .env.example 39└── Makefile 40
这里有一条很实用的边界:
endpoints负责 HTTP 输入输出。schemas保存 Pydantic 请求和响应模型。services放业务规则。repositories封装数据库访问。db/models.py保存 SQLAlchemy ORM 模型。core放跨业务的基础设施配置。
不要让路由函数直接堆几十行 SQL、缓存和权限判断。项目刚开始时似乎省事,三个月后它就会长成一团很有生命力的藤蔓。
三、依赖与配置管理
推荐使用 uv 管理 Python 依赖和锁文件,也可以换成 Poetry。核心原则是提交锁文件,并在构建时严格按锁文件安装。
pyproject.toml
1[project] 2name = "fastapi-project" 3version = "0.1.0" 4requires-python = ">=3.12" 5dependencies = [ 6 "fastapi", 7 "uvicorn[standard]", 8 "sqlalchemy[asyncio]", 9 "asyncpg", 10 "alembic", 11 "redis[hiredis]", 12 "pydantic-settings", 13 "structlog", 14] 15 16[dependency-groups] 17dev = [ 18 "httpx", 19 "pytest", 20 "pytest-asyncio", 21 "ruff", 22 "mypy", 23 "pre-commit", 24] 25 26[tool.ruff] 27line-length = 100 28target-version = "py312" 29 30[tool.pytest.ini_options] 31asyncio_mode = "auto" 32testpaths = ["tests"] 33
正式项目最好锁定依赖版本,不要让生产镜像每次构建都随手拿到不同版本。
app/core/config.py
1from functools import lru_cache 2 3from pydantic import SecretStr 4from pydantic_settings import BaseSettings, SettingsConfigDict 5 6 7class Settings(BaseSettings): 8 app_name: str = "fastapi-project" 9 environment: str = "development" 10 debug: bool = False 11 12 database_url: str 13 redis_url: str 14 15 database_pool_size: int = 10 16 database_max_overflow: int = 10 17 18 secret_key: SecretStr 19 20 model_config = SettingsConfigDict( 21 env_file=".env", 22 env_file_encoding="utf-8", 23 case_sensitive=False, 24 extra="ignore", 25 ) 26 27 28@lru_cache 29def get_settings() -> Settings: 30 return Settings() 31 32 33settings = get_settings() 34
.env.example 只保留示例,不放真实密码。
1ENVIRONMENT=development 2DEBUG=true 3 4DATABASE_URL=postgresql+asyncpg://app:app@postgres:5432/app 5REDIS_URL=redis://redis:6379/0 6 7DATABASE_POOL_SIZE=10 8DATABASE_MAX_OVERFLOW=10 9 10SECRET_KEY=replace-me 11
开发环境使用 .env 没问题,生产环境则应使用 Docker Secrets、Kubernetes Secrets、Vault 或云厂商的密钥服务。秘密写进镜像或 Git 历史后,即使后来删除文件,也不能算真正删除。Docker 官方同样建议通过 secrets 传递密码、证书和令牌等敏感数据。
四、PostgreSQL 集成
FastAPI 是异步框架时,数据库层可以采用 SQLAlchemy 2.x 的异步接口和 asyncpg。每个请求创建一个短生命周期的 AsyncSession,请求结束后归还连接。
app/db/session.py
1from collections.abc import AsyncIterator 2 3from sqlalchemy.ext.asyncio import ( 4 AsyncSession, 5 async_sessionmaker, 6 create_async_engine, 7) 8 9from app.core.config import settings 10 11 12engine = create_async_engine( 13 settings.database_url, 14 pool_size=settings.database_pool_size, 15 max_overflow=settings.database_max_overflow, 16 pool_pre_ping=True, 17 pool_recycle=1800, 18 echo=False, 19) 20 21AsyncSessionFactory = async_sessionmaker( 22 bind=engine, 23 class_=AsyncSession, 24 expire_on_commit=False, 25 autoflush=False, 26) 27 28 29async def get_db_session() -> AsyncIterator[AsyncSession]: 30 async with AsyncSessionFactory() as session: 31 try: 32 yield session 33 except Exception: 34 await session.rollback() 35 raise 36
业务层可以显式决定事务边界。
1from sqlalchemy.ext.asyncio import AsyncSession 2 3from app.db.models import User 4 5 6async def create_user(session: AsyncSession, email: str) -> User: 7 user = User(email=email) 8 session.add(user) 9 10 await session.commit() 11 await session.refresh(user) 12 13 return user 14
连接池配置的关键点
连接池不是越大越好。需要同时考虑:
- API 容器数量
- 每个容器的 worker 数量
- 每个 worker 的连接池上限
- Worker、定时任务和迁移工具占用的连接
- PostgreSQL 的
max_connections
例如,4 个容器、每个容器 2 个进程、每个进程最多使用 20 个连接,理论峰值已经达到 160 个,还没算后台任务和运维连接。
流量较大时,可以在 PostgreSQL 前使用 PgBouncer。若采用事务级池化,需要检查预编译语句、会话变量等功能是否与池化模式兼容。
SQLAlchemy 官方文档强调,Session 和 AsyncSession 是有状态的事务对象,不应被多个并发任务共享;连接在事务结束后会被释放回 Engine 管理的连接池。
五、Redis 集成
现代 redis-py 已经内置 asyncio 支持,无需继续使用已经并入主项目的旧版 aioredis 包。
Redis 客户端适合在应用启动时创建,在应用关闭时统一释放,而不是每个请求都重新建连接。
app/main.py
1from contextlib import asynccontextmanager 2 3import redis.asyncio as redis 4from fastapi import FastAPI, Request 5from sqlalchemy import text 6 7from app.core.config import settings 8from app.db.session import AsyncSessionFactory, engine 9 10 11@asynccontextmanager 12async def lifespan(app: FastAPI): 13 redis_client = redis.from_url( 14 settings.redis_url, 15 encoding="utf-8", 16 decode_responses=True, 17 health_check_interval=30, 18 socket_connect_timeout=3, 19 socket_timeout=3, 20 retry_on_timeout=True, 21 ) 22 23 app.state.redis = redis_client 24 25 yield 26 27 await redis_client.aclose() 28 await engine.dispose() 29 30 31app = FastAPI( 32 title=settings.app_name, 33 lifespan=lifespan, 34) 35 36 37@app.get("/health/live", include_in_schema=False) 38async def liveness(): 39 return {"status": "ok"} 40 41 42@app.get("/health/ready", include_in_schema=False) 43async def readiness(request: Request): 44 async with AsyncSessionFactory() as session: 45 await session.execute(text("SELECT 1")) 46 47 await request.app.state.redis.ping() 48 49 return {"status": "ready"} 50
Redis 依赖注入
1from typing import Annotated 2 3from fastapi import Depends, Request 4from redis.asyncio import Redis 5from sqlalchemy.ext.asyncio import AsyncSession 6 7from app.db.session import get_db_session 8 9 10def get_redis(request: Request) -> Redis: 11 return request.app.state.redis 12 13 14DBSession = Annotated[AsyncSession, Depends(get_db_session)] 15RedisClient = Annotated[Redis, Depends(get_redis)] 16
缓存示例
1import json 2 3from redis.asyncio import Redis 4 5 6async def get_cached_user(redis_client: Redis, user_id: int) -> dict | None: 7 value = await redis_client.get(f"user:{user_id}") 8 9 if value is None: 10 return None 11 12 return json.loads(value) 13 14 15async def cache_user( 16 redis_client: Redis, 17 user_id: int, 18 payload: dict, 19) -> None: 20 await redis_client.set( 21 f"user:{user_id}", 22 json.dumps(payload), 23 ex=300, 24 ) 25
工程中建议提前约定缓存键格式,例如:
1项目名:环境:业务:版本:标识 2myapp:prod:user:v1:123 3
更新数据库后,要删除或更新对应缓存。常见模式是 Cache Aside,读取时先查缓存,未命中再查数据库并回填;写入时先提交数据库事务,再使缓存失效。Redis 的 asyncio 客户端和连接池均由官方 redis-py 提供,并要求在应用退出时显式关闭客户端或连接池。
六、Dockerfile 最佳实践
推荐多阶段构建、非 root 用户、固定 Python 版本和依赖锁文件。不要把开发工具和编译器留在最终镜像里。
1# syntax=docker/dockerfile:1.7 2 3FROM python:3.12-slim AS builder 4 5ENV UV_COMPILE_BYTECODE=1 \ 6 UV_LINK_MODE=copy 7 8WORKDIR /build 9 10COPY /uv /usr/local/bin/uv 11 12COPY pyproject.toml uv.lock ./ 13 14RUN \ 15 uv sync \ 16 --frozen \ 17 --no-dev \ 18 --no-install-project 19 20 21FROM python:3.12-slim AS runtime 22 23ENV PYTHONUNBUFFERED=1 \ 24 PYTHONDONTWRITEBYTECODE=1 \ 25 PATH="/app/.venv/bin:$PATH" 26 27RUN groupadd --system --gid 10001 app \ 28 && useradd --system --uid 10001 --gid app --home-dir /app app 29 30WORKDIR /app 31 32COPY /build/.venv /app/.venv 33COPY app /app/app 34COPY migrations /app/migrations 35COPY alembic.ini /app/alembic.ini 36COPY pyproject.toml /app/pyproject.toml 37 38USER app 39 40EXPOSE 8000 41 42CMD ["uvicorn", "app.main:app", \ 43 "--host", "0.0.0.0", \ 44 "--port", "8000", \ 45 "--proxy-headers"] 46
生产镜像中应避免:
- 使用
latest作为 Python 基础镜像标签 - 以 root 身份运行服务
- 把
.env、测试数据或 Git 目录复制进镜像 - 在容器启动时执行
pip install - 开启
--reload - 把 PostgreSQL 和 Redis 塞进 API 镜像
- 将数据库迁移隐式绑定到每个 API 副本的启动过程
镜像构建还可以进一步固定基础镜像 digest。示例里的 uv:latest 为了便于阅读,严谨的生产配置也应锁定版本或 digest。FastAPI 官方的容器指南同样采用从官方 Python 镜像构建、先复制依赖文件以利用缓存、再复制应用代码的方式。
.dockerignore
1.git 2.github 3.idea 4.vscode 5 6.env 7.env.* 8!.env.example 9 10__pycache__ 11*.py[cod] 12.pytest_cache 13.mypy_cache 14.ruff_cache 15 16.venv 17dist 18build 19htmlcov 20.coverage 21tests 22
七、Docker Compose 开发环境
基础 Compose 文件可以同时描述应用和依赖,但开发专属的源码挂载、热更新最好放进覆盖文件。
compose.yaml
1services: 2 api: 3 build: 4 context: . 5 target: runtime 6 command: 7 - uvicorn 8 - app.main:app 9 - --host=0.0.0.0 10 - --port=8000 11 - --proxy-headers 12 env_file: 13 - .env 14 ports: 15 - "8000:8000" 16 depends_on: 17 migrate: 18 condition: service_completed_successfully 19 redis: 20 condition: service_healthy 21 restart: unless-stopped 22 init: true 23 stop_grace_period: 30s 24 networks: 25 - backend 26 27 migrate: 28 build: 29 context: . 30 target: runtime 31 command: ["alembic", "upgrade", "head"] 32 env_file: 33 - .env 34 depends_on: 35 postgres: 36 condition: service_healthy 37 restart: "no" 38 networks: 39 - backend 40 41 postgres: 42 image: postgres:17 43 environment: 44 POSTGRES_DB: app 45 POSTGRES_USER: app 46 POSTGRES_PASSWORD: app 47 volumes: 48 - postgres_data:/var/lib/postgresql/data 49 healthcheck: 50 test: ["CMD-SHELL", "pg_isready -U app -d app"] 51 interval: 5s 52 timeout: 3s 53 retries: 20 54 start_period: 5s 55 restart: unless-stopped 56 networks: 57 - backend 58 59 redis: 60 image: redis:8 61 command: 62 - redis-server 63 - --appendonly 64 - "yes" 65 - --maxmemory 66 - 256mb 67 - --maxmemory-policy 68 - allkeys-lru 69 volumes: 70 - redis_data:/data 71 healthcheck: 72 test: ["CMD", "redis-cli", "ping"] 73 interval: 5s 74 timeout: 3s 75 retries: 20 76 restart: unless-stopped 77 networks: 78 - backend 79 80volumes: 81 postgres_data: 82 redis_data: 83 84networks: 85 backend: 86
这里没有将 PostgreSQL 和 Redis 端口映射到宿主机,因为 API 可以通过 Compose 内部网络访问它们。确实需要从宿主机连接数据库时,再在开发覆盖文件中开放端口。
compose.dev.yaml
1services: 2 api: 3 command: 4 - uvicorn 5 - app.main:app 6 - --host=0.0.0.0 7 - --port=8000 8 - --reload 9 volumes: 10 - ./app:/app/app 11 - ./tests:/app/tests 12 environment: 13 DEBUG: "true" 14 15 postgres: 16 ports: 17 - "127.0.0.1:5432:5432" 18 19 redis: 20 ports: 21 - "127.0.0.1:6379:6379" 22
运行命令:
1docker compose -f compose.yaml -f compose.dev.yaml up --build 2
depends_on 只表达依赖关系还不够,数据库进程已经启动并不意味着它已经能接受连接。配合 healthcheck 和 condition: service_healthy,Compose 才会等待依赖达到健康状态。Docker 官方文档对此有明确说明。
八、数据库迁移
数据库结构变化应由 Alembic 管理,不要在 FastAPI 启动事件中调用 metadata.create_all() 代替正式迁移。
初始化:
1docker compose run --rm api alembic init migrations 2
生成迁移:
1docker compose run --rm api \ 2 alembic revision --autogenerate -m "create users table" 3
应用迁移:
1docker compose run --rm api alembic upgrade head 2
生产环境建议将迁移作为部署流水线中的独立任务。不建议每个 API 容器启动时都自动迁移,因为多个副本可能同时修改数据库。
对于删除列、修改字段类型一类破坏性变更,可采用兼容式迁移:
- 新增字段,但暂不删除旧字段。
- 应用同时兼容新旧结构。
- 回填历史数据。
- 新版本只使用新字段。
- 确认无旧版本运行后,再删除旧字段。
这类 expand-and-contract 流程虽然多走两步,却能显著降低滚动发布期间的故障风险。Alembic 是 SQLAlchemy 官方生态中的迁移工具,支持迁移脚本、版本链以及基于 ORM 元数据的自动差异生成。
九、健康检查、日志与监控
健康检查最好拆成两个接口。
存活检查
/health/live 只判断进程是否正常响应,不访问外部依赖。若失败,编排平台可以重启容器。
就绪检查
/health/ready 检查 PostgreSQL、Redis 等关键依赖。失败时停止接收新流量,但不一定立即重启容器。
不要把复杂业务查询塞进健康检查,否则监控本身也可能变成数据库压力来源。
日志建议
- 日志写入标准输出和标准错误,不写容器内文件。
- 生产环境采用 JSON 结构化日志。
- 每个请求携带
request_id或trace_id。 - 避免记录密码、令牌、Cookie 和完整个人信息。
- 记录请求耗时、状态码和异常类型。
- 接入 OpenTelemetry、Prometheus、Grafana 或云厂商监控。
推荐至少观察这些指标:
- 请求量、错误率和延迟分位数
- PostgreSQL 活跃连接数与连接等待时间
- 慢查询数量和事务回滚率
- Redis 命中率、内存占用、过期和驱逐数量
- 容器 CPU、内存和重启次数
- 后台任务积压量
十、测试与持续集成
测试应分为三个层次。
| 层次 | 内容 | 是否需要真实依赖 |
|---|---|---|
| 单元测试 | Service、纯函数、权限规则 | 通常不需要 |
| 集成测试 | Repository、SQL、Redis 行为 | 需要 |
| API 测试 | 路由、认证、事务与响应 | 通常需要 |
数据库相关测试尽量使用真实 PostgreSQL,而不是用 SQLite 替代。两者在字段类型、约束、JSON、事务和 SQL 语法上的差异,可能让测试产生虚假的安全感。
CI 流程可以设计为:
1Ruff 检查 2 ↓ 3类型检查 4 ↓ 5单元测试 6 ↓ 7启动 PostgreSQL 和 Redis 8 ↓ 9执行 Alembic 迁移 10 ↓ 11集成测试 12 ↓ 13构建镜像 14 ↓ 15镜像漏洞扫描 16 ↓ 17推送镜像仓库 18
镜像标签建议同时包含 Git 提交号,例如:
1registry.example.com/myapp/api:git-a1b2c3d 2
部署时使用不可变标签或 digest,不依赖 latest。
十一、生产环境需要调整的地方
开发 Compose 不能原封不动搬进生产环境。正式部署至少应完成下面这些调整。
- PostgreSQL 和 Redis 优先使用托管服务,或建立成熟的备份、恢复和高可用机制。
- 不对公网暴露数据库与 Redis 端口。
- 使用 secrets 管理密码和证书。
- 配置 CPU、内存限制以及合理的重启策略。
- 在入口层终止 TLS,并配置可信代理。
- API 镜像以非 root 用户运行,并尽量使用只读文件系统。
- 定期执行 PostgreSQL 备份和恢复演练。
- Redis 若只承担缓存,可以接受数据丢失;若承担队列或锁,要明确持久化和故障语义。
- 后台任务单独运行,不在请求处理函数中执行长时间计算。
- 对上传文件使用对象存储,不依赖容器本地目录。
- 在容器收到
SIGTERM后停止接收新请求,并留出优雅关闭时间。
如果运行在 Kubernetes、ECS 或类似平台,通常采用一个容器一个 Uvicorn worker,通过增加 Pod 或任务数扩容。若只是单台 Linux 服务器,可以在一个容器中设置少量 worker,但必须重新核算数据库连接池。
十二、一套实用的落地顺序
别一上来就铺满 Kubernetes、消息队列和十几个微服务。对多数新项目,更稳妥的演进路径是:
- 建立上述目录结构和依赖锁文件。
- 用 Compose 启动 FastAPI、PostgreSQL 和 Redis。
- 接入 SQLAlchemy 异步会话与 Alembic。
- 通过 lifespan 管理 Redis 和数据库资源。
- 增加存活与就绪检查。
- 配置 Ruff、pytest 和 CI。
- 构建非 root、多阶段生产镜像。
- 接入结构化日志与监控。
- 有明确性能数据后,再调整 worker、连接池和缓存策略。
- 业务确实需要时,再引入 Celery、Dramatiq、Arq 或其他任务系统。
这套方案的核心并不花哨——镜像可复现、配置与代码分离、依赖有健康检查、事务边界清晰、迁移独立执行、开发和生产环境明确分层。这些基础打牢之后,项目即使逐渐变大,也不会很快陷入只能祈祷部署成功的阶段。
参考资料
- FastAPI Documentation, FastAPI in Containers – Docker
fastapi.tiangolo.com/deployment/… - Docker Documentation, Control startup and shutdown order in Compose;Manage sensitive data with Docker secrets
docs.docker.com/compose/how…
docs.docker.com/engine/swar… - SQLAlchemy Documentation, Asynchronous I/O
docs.sqlalchemy.org/en/latest/o… - SQLAlchemy Documentation, Connection Pooling;Session Basics
docs.sqlalchemy.org/en/latest/c…
docs.sqlalchemy.org/en/latest/o… - redis-py Documentation, Asyncio Examples
redis.readthedocs.io/en/stable/e… - Alembic Documentation, SQLAlchemy Alembic Documentation
alembic.sqlalchemy.org/en/latest/
《Linux + Docker + FastAPI 工程化实践指南》 是转载文章,点击查看原文。
