fastapi-enterprise

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

FastAPI Enterprise Development - Modular Architecture

FastAPI企业级开发 - 模块化架构

Enterprise-grade FastAPI with modular architecture: each business domain is an independent module with own database tables, cache namespace, routes, and migrations. Modern Python tooling (UV + pyproject.toml) for fast dependency management.
具备模块化架构的企业级FastAPI:每个业务域都是独立模块,拥有专属的数据库表、缓存命名空间、路由和迁移机制。采用现代Python工具链(UV + pyproject.toml)实现快速依赖管理。

Quick Start

快速开始

Initialize Project

初始化项目

bash
undefined
bash
undefined

Create modular FastAPI project with UV

使用UV创建模块化FastAPI项目

python scripts/init_project.py --name my_api --auth keycloak --with-example-module
python scripts/init_project.py --name my_api --auth keycloak --with-example-module

Result:

生成结果:

my_api/ ← Project name subdirectory

my_api/ ← 项目名称子目录

├── src/

├── src/

│ ├── app.py ← Main application

│ ├── app.py ← 主应用文件

│ ├── core/ ← Shared infrastructure (logging, DB, cache, module loader)

│ ├── core/ ← 共享基础设施(日志、数据库、缓存、模块加载器)

│ ├── middleware/ ← Conversation tracking, auth

│ ├── middleware/ ← 会话追踪、认证中间件

│ └── routes/ ← Core routes (health, metrics)

│ └── routes/ ← 核心路由(健康检查、指标)

├── modules/ ← Independent modules

├── modules/ ← 独立模块目录

│ └── users/ ← Example: own routes, models, cache, migrations

│ └── users/ ← 示例模块:拥有专属路由、模型、缓存、迁移

├── pyproject.toml ← UV configuration

├── pyproject.toml ← UV配置文件

├── config/ ← YAML configurations

├── config/ ← YAML配置文件目录

└── scripts/ ← Automation (create_module.py)

└── scripts/ ← 自动化脚本目录(create_module.py等)

Install and run

安装依赖并运行

cd my_api uv sync # Install dependencies (10-100x faster than pip) cp .env.example .env uv run uvicorn src.app:app --reload
undefined
cd my_api uv sync # 安装依赖(速度比pip快10-100倍) cp .env.example .env uv run uvicorn src.app:app --reload
undefined

Create Independent Module

创建独立模块

bash
undefined
bash
undefined

Generate new module with complete structure

生成完整结构的新模块

uv run python scripts/create_module.py --name orders
uv run python scripts/create_module.py --name orders

Auto-creates:

自动创建以下内容:

modules/orders/

modules/orders/

├── init.py ← Router export (auto-discovered)

├── init.py ← 路由导出(自动发现)

├── routes/ ← /api/v1/orders endpoints

├── routes/ ← /api/v1/orders 端点

├── models/ ← SQLAlchemy models (own tables)

├── models/ ← SQLAlchemy模型(专属数据表)

├── schemas/ ← Pydantic validation

├── schemas/ ← Pydantic验证模式

├── services/ ← Business logic

├── services/ ← 业务逻辑层

├── cache/ ← Module-specific cache (orders:* prefix)

├── cache/ ← 模块专属缓存(orders:* 前缀)

├── enums/ ← Module-specific enums

├── enums/ ← 模块专属枚举

└── alembic/ ← Module-specific migrations

└── alembic/ ← 模块专属迁移脚本

undefined
undefined

Minimal Modular App

最小化模块化应用示例

python
undefined
python
undefined

src/app.py

src/app.py

from fastapi import FastAPI from src.core.config import settings from src.core.logging import configure_logging from src.core.cache import cache_manager from src.core.module_loader import discover_modules from src.middleware.conversation_middleware import ConversationMiddleware
from fastapi import FastAPI from src.core.config import settings from src.core.logging import configure_logging from src.core.cache import cache_manager from src.core.module_loader import discover_modules from src.middleware.conversation_middleware import ConversationMiddleware

Configure logging (JSON/colored)

配置日志(JSON格式/彩色输出)

configure_logging(environment=settings.ENVIRONMENT)
app = FastAPI(title=settings.PROJECT_NAME) app.add_middleware(ConversationMiddleware) # UUID tracking
@app.on_event("startup") async def startup(): await cache_manager.connect() # Redis or memory discover_modules(app) # Auto-register all modules
@app.get("/") async def root(): return {"message": "Welcome"}
undefined
configure_logging(environment=settings.ENVIRONMENT)
app = FastAPI(title=settings.PROJECT_NAME) app.add_middleware(ConversationMiddleware) # UUID会话追踪
@app.on_event("startup") async def startup(): await cache_manager.connect() # 连接Redis或内存缓存 discover_modules(app) # 自动注册所有模块
@app.get("/") async def root(): return {"message": "欢迎使用"}
undefined

Core Capabilities

核心能力

1. Modular Architecture ⭐

1. 模块化架构 ⭐

  • Independent Modules: Each module owns DB tables, cache keys, routes
  • Auto-Discovery: Modules in
    modules/
    automatically registered
  • Service-Layer Communication: Inter-module calls via services (no direct model imports)
  • Per-Module Migrations: Each module has own Alembic history
  • Enforced Boundaries:
    modules.users.services.UserService
    ✅ |
    from modules.users.models import User
Module Patterns Guide
  • 独立模块: 每个模块拥有专属数据表、缓存键、路由
  • 自动发现:
    modules/
    目录下的模块会自动注册
  • 服务层通信: 模块间通过服务层调用(禁止直接导入模型)
  • 基于模块的迁移: 每个模块拥有独立的Alembic迁移历史
  • 强制边界约束:
    modules.users.services.UserService
    ✅ |
    from modules.users.models import User
模块模式指南

2. Modern Python Tooling (UV + pyproject.toml) ⭐

2. 现代Python工具链(UV + pyproject.toml)⭐

  • UV: 10-100x faster than pip, Rust-based package manager
  • pyproject.toml: PEP 518/621 standard, replaces requirements.txt
  • uv.lock: Reproducible builds across environments
  • Project Subdirectory: All files in
    my_api/
    (clean workspace)
  • Export:
    uv export --format requirements-txt > requirements.txt
    when needed
Project Structure
  • UV: 基于Rust的包管理器,速度比pip快10-100倍
  • pyproject.toml: 遵循PEP 518/621标准,替代requirements.txt
  • uv.lock: 跨环境可复现的构建锁文件
  • 项目子目录: 所有文件位于
    my_api/
    目录下(工作区整洁)
  • 导出兼容: 必要时可执行
    uv export --format requirements-txt > requirements.txt
    生成requirements.txt
项目结构说明

3. Flexible Caching ⭐

3. 灵活的缓存机制 ⭐

  • Redis (production): Distributed, persistent cache
  • Memory (development): Zero dependencies, instant setup
  • Module Namespaces:
    users:123
    ,
    orders:456
    (no collisions)
  • Cache Decorators:
    @cached(prefix="products", ttl=1800)
  • Invalidation: Time-based (TTL), event-based (on update), pattern-based (
    users:*
    )
Cache Patterns
  • Redis(生产环境): 分布式持久化缓存
  • 内存缓存(开发环境): 零依赖,即时搭建
  • 模块命名空间:
    users:123
    ,
    orders:456
    (避免键冲突)
  • 缓存装饰器:
    @cached(prefix="products", ttl=1800)
  • 失效策略: 基于时间(TTL)、基于事件(更新时触发)、基于模式(如
    users:*
缓存模式指南

4. Structured Logging (structlog)

4. 结构化日志(structlog)

  • JSON (production): Structured logs for aggregation (ELK, Splunk)
  • Colored Console (development): Human-readable with syntax highlighting
  • Conversation ID: UUID in all logs from
    ConversationMiddleware
  • Request/Response: Automatic logging with duration, status code
  • External APIs: Conversation ID propagated to downstream services
Logging Patterns
  • JSON格式(生产环境): 结构化日志便于聚合分析(ELK、Splunk)
  • 彩色控制台输出(开发环境): 人类可读的语法高亮日志
  • 会话ID: 所有日志包含来自
    ConversationMiddleware
    的UUID会话ID
  • 请求/响应自动日志: 自动记录请求时长、状态码等信息
  • 外部API追踪: 会话ID会自动传递给下游服务
日志模式指南

5. Type-Safe Configuration

5. 类型安全的配置管理

  • pydantic-settings: OS environment variables with type validation
  • YAML:
    config/development.yml
    and
    config/production.yml
  • Variable Substitution:
    ${DATABASE_PASSWORD}
    from environment
  • CONFIG_ENV Switcher: Toggle between dev/prod configs
Configuration Patterns
  • pydantic-settings: 对操作系统环境变量进行类型验证
  • YAML配置:
    config/development.yml
    config/production.yml
  • 变量替换: 从环境变量中读取
    ${DATABASE_PASSWORD}
    等值
  • CONFIG_ENV切换器: 快速切换开发/生产配置
配置模式指南

6. Async Database (SQLAlchemy 2.0)

6. 异步数据库(SQLAlchemy 2.0)

  • Async: Full asyncio support with
    asyncpg
    driver
  • PostgreSQL-Optimized: JSONB, arrays, full-text search
  • Per-Module Tables: Each module creates own tables
  • Central Session:
    src/core/db.py
    session factory used by all modules and Alembic
Database Patterns
  • 异步支持: 完整的asyncio支持,搭配
    asyncpg
    驱动
  • PostgreSQL优化: 支持JSONB、数组、全文搜索
  • 模块专属表: 每个模块创建专属数据表
  • 集中式会话: 所有模块和Alembic使用
    src/core/db.py
    中的会话工厂
数据库模式指南

7. Per-Module Migrations (Alembic)

7. 基于模块的迁移(Alembic)

  • Independent Histories:
    modules/users/alembic/
    ,
    modules/orders/alembic/
  • Async Template: Configured for async SQLAlchemy
  • Auto-Generation:
    alembic revision --autogenerate
  • Run All:
    python scripts/run_migrations.py --all
Alembic Setup
  • 独立迁移历史:
    modules/users/alembic/
    ,
    modules/orders/alembic/
  • 异步模板: 适配异步SQLAlchemy的配置模板
  • 自动生成: 执行
    alembic revision --autogenerate
    自动生成迁移
  • 批量运行: 执行
    python scripts/run_migrations.py --all
    运行所有模块的迁移
Alembic配置指南

8. Central HTTP Client (httpx)

8. 集中式HTTP客户端(httpx)

  • Async: Connection pooling, timeouts, retries
  • Conversation ID: Auto-propagated to
    X-Conversation-ID
    header
  • YAML Configs: Base URLs from
    config/*.yml
  • Request Logging: All external API calls logged
HTTPx Patterns
  • 异步支持: 连接池、超时、重试机制
  • 会话ID传递: 自动将会话ID添加到
    X-Conversation-ID
    请求头
  • YAML配置: 从
    config/*.yml
    读取基础URL
  • 请求日志: 所有外部API调用都会被记录
HTTPx模式指南

9. Observability

9. 可观测性

  • Conversation Tracking: UUID from header/cookie, appears in all logs
  • Prometheus Metrics:
    /metrics
    endpoint (request count, latency, errors)
  • OpenTelemetry (optional): Distributed tracing (Jaeger, Zipkin)
  • Health Check:
    /health
    with DB connectivity status
Observability Patterns
  • 会话追踪: 从请求头/Cookie获取UUID,所有日志均包含该ID
  • Prometheus指标:
    /metrics
    端点(请求计数、延迟、错误数)
  • OpenTelemetry(可选): 分布式追踪(支持Jaeger、Zipkin)
  • 健康检查:
    /health
    端点,包含数据库连接状态
可观测性模式指南

10. Authentication (Conditional)

10. 可配置认证机制

AI MUST ASK: "Keycloak authentication (JWT from external IdP) or app-based RBAC (database-backed roles)?"
  • Keycloak: JWT validation, role extraction, Keycloak OpenID integration
  • App-based: Custom JWT, database roles/permissions, user management
  • Middleware: Token validation,
    request.state.current_user
    injection
  • Role Decorators:
    @requires_role("admin")
Authentication Patterns
必须确认: "需要Keycloak认证(来自外部身份提供商的JWT)还是基于应用的RBAC(数据库存储角色)?"
  • Keycloak: JWT验证、角色提取、Keycloak OpenID集成
  • 基于应用: 自定义JWT、数据库存储角色/权限、用户管理
  • 中间件: 令牌验证、注入
    request.state.current_user
  • 角色装饰器:
    @requires_role("admin")
认证模式指南

Automation Scripts

自动化脚本

init_project.py

init_project.py

Initialize complete project with modular structure:
bash
python scripts/init_project.py --name my_api --auth keycloak --with-example-module
初始化完整的模块化项目:
bash
python scripts/init_project.py --name my_api --auth keycloak --with-example-module

create_module.py

create_module.py

Generate new independent module:
bash
uv run python scripts/create_module.py --name products
生成新的独立模块:
bash
uv run python scripts/create_module.py --name products

create_endpoint.py

create_endpoint.py

Add endpoint to existing module:
bash
uv run python scripts/create_endpoint.py --module users --service auth --method post --path login
为现有模块添加端点:
bash
uv run python scripts/create_endpoint.py --module users --service auth --method post --path login

create_model.py

create_model.py

Create SQLAlchemy model + Alembic migration:
bash
uv run python scripts/create_model.py --module products --name Product
创建SQLAlchemy模型并生成Alembic迁移:
bash
uv run python scripts/create_model.py --module products --name Product

validate_structure.py

validate_structure.py

Check project compliance:
bash
uv run python scripts/validate_structure.py
检查项目结构合规性:
bash
uv run python scripts/validate_structure.py

Module Independence Rules

模块独立性规则

✅ DO:

✅ 允许操作:

  • Service Layer Communication:
    from modules.users.services.user_service import UserService
  • Module Namespaces:
    users:cache:123
    ,
    orders:cache:456
  • Own Tables: Each module creates own database tables
  • Own Migrations: Separate Alembic history per module
  • 服务层通信:
    from modules.users.services.user_service import UserService
  • 模块命名空间:
    users:cache:123
    ,
    orders:cache:456
  • 专属数据表: 每个模块创建专属数据库表
  • 独立迁移: 每个模块拥有单独的Alembic迁移历史

❌ DON'T:

❌ 禁止操作:

  • Direct Model Imports:
    from modules.users.models.user import User
    (use service instead)
  • Shared Tables: No
    ForeignKey("users.id")
    across modules (store as regular int)
  • Cross-Module Cache: Don't access other module's cache directly
Clean Code Standards
  • 直接导入模型:
    from modules.users.models.user import User
    (请使用服务层替代)
  • 共享数据表: 禁止跨模块使用
    ForeignKey("users.id")
    (请存储为普通整数)
  • 跨模块访问缓存: 禁止直接访问其他模块的缓存
整洁代码标准

Resources

资源

Reference Documentation

参考文档

  • Project Structure - Directory layout, UV setup
  • Module Patterns - Independent module development
  • Cache Patterns - Redis/memory caching strategies
  • Routing Patterns - Auto-discovery, module loader
  • Logging Patterns - Structlog configuration
  • Configuration Patterns - pydantic-settings, YAML
  • Database Patterns - Async SQLAlchemy, CRUD
  • Alembic Setup - Per-module migrations
  • HTTPx Patterns - Central HTTP client
  • Observability Patterns - Conversation tracking, metrics
  • Authentication Patterns - Keycloak vs app-based
  • Clean Code Standards - DDD, no duplication
  • Error Handling Workflow - Systematic debugging
  • 项目结构 - 目录布局、UV配置
  • 模块模式 - 独立模块开发指南
  • 缓存模式 - Redis/内存缓存策略
  • 路由模式 - 自动发现、模块加载器
  • 日志模式 - Structlog配置
  • 配置模式 - pydantic-settings、YAML配置
  • 数据库模式 - 异步SQLAlchemy、CRUD操作
  • Alembic配置 - 基于模块的迁移
  • HTTPx模式 - 集中式HTTP客户端
  • 可观测性模式 - 会话追踪、指标监控
  • 认证模式 - Keycloak vs 基于应用的认证
  • 整洁代码标准 - DDD、无重复代码
  • 错误处理流程 - 系统化调试

Automation Scripts

自动化脚本

  • scripts/init_project.py
    - Initialize complete project
  • scripts/create_module.py
    - Generate new module
  • scripts/create_endpoint.py
    - Add endpoint to module
  • scripts/create_model.py
    - Create model + migration
  • scripts/validate_structure.py
    - Check project compliance
  • scripts/init_project.py
    - 初始化完整项目
  • scripts/create_module.py
    - 生成新模块
  • scripts/create_endpoint.py
    - 为模块添加端点
  • scripts/create_model.py
    - 创建模型并生成迁移
  • scripts/validate_structure.py
    - 检查项目合规性

Examples

示例

  • Complete Usage Examples - 6 full examples (module creation, cache usage, auth, metrics, etc.)
  • 完整使用示例 - 6个完整示例(模块创建、缓存使用、认证、指标等)

License

许可证

MIT License - See LICENSE.txt for complete terms.
MIT许可证 - 详见LICENSE.txt获取完整条款。