python-patterns

Python 模式

2025 年的 Python 开发原则与决策。 学会思考，而不是死记模式。

适用场景

在制定 Python 架构决策、选择框架、设计异步模式或规划 Python 项目结构时，使用此技能。

---

⚠️ 如何使用此技能

此技能讲解的是决策原则，而不是可直接复制的固定代码。

• 不确定时，询问用户对框架的偏好
• 根据上下文决定使用异步还是同步
• 不要每次都默认使用同一个框架

---

1. 框架选择（2025）

决策树

What are you building?
│
├── API-first / Microservices
│   └── FastAPI (async, modern, fast)
│
├── Full-stack web / CMS / Admin
│   └── Django (batteries-included)
│
├── Simple / Script / Learning
│   └── Flask (minimal, flexible)
│
├── AI/ML API serving
│   └── FastAPI (Pydantic, async, uvicorn)
│
└── Background workers
    └── Celery + any framework

比较原则

因素	FastAPI	Django	Flask
最适合	API、微服务	全栈、CMS	简单项目、学习
异步	支持 ASGI	支持 ASGI；检查 ORM/事务限制	异步视图需安装 async extra；仍受 WSGI 限制
管理后台	手动实现	内置	通过扩展实现
ORM	自行选择	Django ORM	自行选择
学习曲线	低	中	低

需要提出的选择问题：

1. 这是仅含 API 的项目还是全栈项目？
2. 是否需要管理界面？
3. 团队是否熟悉异步编程？
4. 是否有现成的基础设施？

---

2. 异步还是同步

何时使用异步

async def is better when:
├── I/O-bound operations (database, HTTP, file)
├── Many concurrent connections
├── Real-time features
├── Microservices communication
└── FastAPI/Starlette/Django ASGI

def (sync) is better when:
├── CPU-bound operations
├── Simple scripts
├── Legacy codebase
├── Team unfamiliar with async
└── Blocking libraries (no async version)

黄金法则

I/O-bound → async (waiting for external)
CPU-bound → sync + multiprocessing (computing)

Don't:
├── Mix sync and async carelessly
├── Use sync libraries in async code
└── Force async for CPU work

异步库选择

需求	异步库
HTTP 客户端	httpx
PostgreSQL	asyncpg
Redis	aioredis / redis-py async
文件 I/O	aiofiles
数据库 ORM	SQLAlchemy 2.0 async、Tortoise

---

3. 类型提示策略

何时添加类型

Always type:
├── Function parameters
├── Return types
├── Class attributes
├── Public APIs

Can skip:
├── Local variables (let inference work)
├── One-off scripts
├── Tests (usually)

常见类型模式

# These are patterns, understand them:

# Optional → might be None
from typing import Optional
def find_user(id: int) -> Optional[User]: ...

# Union → one of multiple types
def process(data: str | dict) -> None: ...

# Generic collections
def get_items() -> list[Item]: ...
def get_mapping() -> dict[str, int]: ...

# Callable
from typing import Callable
def apply(fn: Callable[[int], str]) -> str: ...

使用 Pydantic 进行验证

When to use Pydantic:
├── API request/response models
├── Configuration/settings
├── Data validation
├── Serialization

Benefits:
├── Runtime validation
├── Auto-generated JSON schema
├── Works with FastAPI natively
└── Clear error messages

---

4. 项目结构原则

结构选择

Small project / Script:
├── main.py
├── utils.py
└── requirements.txt

Medium API:
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models/
│   ├── routes/
│   ├── services/
│   └── schemas/
├── tests/
└── pyproject.toml

Large application:
├── src/
│   └── myapp/
│       ├── core/
│       ├── api/
│       ├── services/
│       ├── models/
│       └── ...
├── tests/
└── pyproject.toml

FastAPI 结构原则

Organize by feature or layer:

By layer:
├── routes/ (API endpoints)
├── services/ (business logic)
├── models/ (database models)
├── schemas/ (Pydantic models)
└── dependencies/ (shared deps)

By feature:
├── users/
│   ├── routes.py
│   ├── service.py
│   └── schemas.py
└── products/
    └── ...

---

5. Django 原则（2025）

Django 异步（Django 5.0+）

Django supports async:
├── Async views
├── Async middleware
├── Async ORM (limited)
└── ASGI deployment

When to use async in Django:
├── External API calls
├── WebSocket (Channels)
├── High-concurrency views
└── Background task triggering

Django 最佳实践

Model design:
├── Fat models, thin views
├── Use managers for common queries
├── Abstract base classes for shared fields

Views:
├── Class-based for complex CRUD
├── Function-based for simple endpoints
├── Use viewsets with DRF

Queries:
├── select_related() for FKs
├── prefetch_related() for M2M
├── Avoid N+1 queries
└── Use .only() for specific fields

---

6. FastAPI 原则

FastAPI 中的 async def 与 def

Use async def when:
├── Using async database drivers
├── Making async HTTP calls
├── I/O-bound operations
└── Want to handle concurrency

Use def when:
├── Blocking operations
├── Sync database drivers
├── CPU-bound work
└── FastAPI runs in threadpool automatically

依赖注入

Use dependencies for:
├── Database sessions
├── Current user / Auth
├── Configuration
├── Shared resources

Benefits:
├── Testability (mock dependencies)
├── Clean separation
├── Automatic cleanup (yield)

Pydantic v2 集成

# FastAPI + Pydantic are tightly integrated:

# Request validation
@app.post("/users")
async def create(user: UserCreate) -> UserResponse:
    # user is already validated
    ...

# Response serialization
# Return type becomes response schema

---

7. 后台任务

选择指南

方案	最适合
BackgroundTasks	简单的进程内任务
Celery	分布式、复杂工作流
ARQ	基于 Redis 的异步任务
RQ	简单的 Redis 队列
Dramatiq	基于 Actor，较 Celery 更简单

何时使用各方案

FastAPI BackgroundTasks:
├── Quick operations
├── No persistence needed
├── Fire-and-forget
└── Same process

Celery/ARQ:
├── Long-running tasks
├── Need retry logic
├── Distributed workers
├── Persistent queue
└── Complex workflows

---

8. 错误处理原则

异常策略

In FastAPI:
├── Create custom exception classes
├── Register exception handlers
├── Return consistent error format
└── Log without exposing internals

Pattern:
├── Raise domain exceptions in services
├── Catch and transform in handlers
└── Client gets clean error response

错误响应理念

Include:
├── Error code (programmatic)
├── Message (human readable)
├── Details (field-level when applicable)
└── NOT stack traces (security)

---

9. 测试原则

测试策略

类型	用途	工具
单元测试	业务逻辑	pytest
集成测试	API 端点	pytest + httpx/TestClient
端到端测试	完整工作流	pytest + DB

异步测试

# Use pytest-asyncio for async tests

import pytest
from httpx import ASGITransport, AsyncClient

@pytest.mark.asyncio
async def test_endpoint():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get("/users")
        assert response.status_code == 200

Fixture 策略

Common fixtures:
├── db_session → Database connection
├── client → Test client
├── authenticated_user → User with token
└── sample_data → Test data setup

---

ASGI 测试示例需要应用 fixture，且不会自动管理启动/关闭；必要时请使用项目的 lifespan fixture/manager。原生异步视图并不会让阻塞库或 CPU 密集型工作变为非阻塞。请参阅 HTTPX ASGI transport。

10. 决策检查清单

实施前：

• 是否询问了用户对框架的偏好？
• 是否根据当前上下文选择了框架？（而非直接使用默认框架）
• 是否决定使用异步还是同步？
• 是否规划了类型提示策略？
• 是否定义了项目结构？
• 是否规划了错误处理？
• 是否考虑了后台任务？

---

11. 应避免的反模式

❌ 不要：

• 简单 API 默认使用 Django（FastAPI 可能更合适）
• 在异步代码中使用同步库
• 不为公共 API 添加类型提示
• 将业务逻辑放在路由/视图中
• 忽略 N+1 查询
• 不谨慎地混用异步与同步代码

✅ 应做：

• 根据上下文选择框架
• 询问异步需求
• 使用 Pydantic 进行验证
• 分离关注点（路由 → 服务 → 仓库）
• 测试关键路径

---

请记住：Python 模式的选择取决于你的具体场景。不要照搬代码，而要思考什么最适合你的应用。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为环境特定验证、测试或专家审核的替代品。
• 如果缺少必要输入、权限、安全边界或成功标准，请停止并请求澄清。