python-backend

Python 后端

使用 asyncio、FastAPI、SQLAlchemy 2.0 和连接池构建生产级 Python 后端的模式。每个类别都有单独的规则文件，位于 rules/ 中，按需加载。

快速参考

类别	规则	影响	适用场景
Asyncio	3	HIGH	TaskGroup、结构化并发、取消处理
FastAPI	3	HIGH	依赖项、中间件、后台任务
SQLAlchemy	3	HIGH	异步会话、关系、迁移
Pooling	3	MEDIUM	数据库连接池、HTTP 会话、调优

共计：4 个类别下有 12 条规则。 从精简文件中恢复的项目决策保存在 references/ork-delta.md 中；供应商资料以链接形式提供，不在此处重述（参见上游覆盖范围（不重述））。

快速开始

# FastAPI + SQLAlchemy async session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# Reusable dependency alias (FastAPI's recommended Annotated form)
SessionDep = Annotated[AsyncSession, Depends(get_db)]

@router.get("/users/{user_id}")
async def get_user(user_id: UUID, db: SessionDep):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# Asyncio TaskGroup with timeout
async def fetch_all(urls: list[str]) -> list[dict]:
    async with asyncio.timeout(30):
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_url(url)) for url in urls]
    return [t.result() for t in tasks]

Asyncio

使用结构化并发、TaskGroup 和 Python 3.11+ 功能的现代 Python asyncio 模式。

关键模式

• TaskGroup 以结构化并发和自动取消取代 gather()
• asyncio.timeout() 上下文管理器，用于组合超时
• Semaphore 用于限制并发（对 HTTP 请求进行速率限制）
• 使用 ExceptionGroup 和 except* 处理多个任务失败
• asyncio.to_thread() 用于将同步代码桥接到异步代码

关键决策

决策	建议
任务启动	使用 TaskGroup，而非 gather()
超时	使用 asyncio.timeout() 上下文管理器
并发限制	使用 asyncio.Semaphore
同步桥接	使用 asyncio.to_thread()
取消	始终重新抛出 CancelledError

FastAPI

适用于生产环境的 FastAPI 生命周期、依赖项、中间件和设置模式。

关键模式

• 使用 asynccontextmanager 管理启动和关闭期间的资源
• 使用基于类的服务和 Depends() 进行依赖注入
• 中间件栈：CORS -> RequestID -> Timing -> Logging
• 使用 .env 和字段验证的 Pydantic Settings
• 配置异常处理器，使其返回 RFC 9457 Problem Details 响应体（响应体格式本身由 ork:api-design 定义）

关键决策

决策	建议
生命周期	使用 asynccontextmanager（而非 events）
依赖项	使用基于类的服务和 DI
设置	使用带 .env 的 Pydantic Settings
响应	使用 ORJSONResponse 提升性能
健康检查	检查所有关键依赖项

SQLAlchemy

使用 SQLAlchemy 2.0、AsyncSession 和 FastAPI 集成的异步数据库模式。

关键模式

• 每个请求使用一个 AsyncSession，并设置 expire_on_commit=False
• 在关系上设置 lazy="raise"，防止意外的 N+1 查询
• 使用 selectinload 预加载集合
• 使用仓储模式实现通用异步 CRUD
• 批量插入按每批 1000-10000 行分块，以便管理内存

关键决策

决策	建议
会话作用域	每个请求使用一个 AsyncSession
延迟加载	使用 lazy="raise" + 显式加载
预加载	对集合使用 selectinload
expire_on_commit	False（防止延迟加载错误）
连接池	设置 pool_pre_ping=True

Pooling

适用于高性能异步 Python 应用的数据库和 HTTP 连接池。

关键模式

• 使用带有 pool_size、max_overflow、pool_pre_ping 的 SQLAlchemy 连接池
• 使用带有 min_size/max_size 和连接生命周期管理的 直接 asyncpg 连接池
• 使用带有 TCPConnector 限制和 DNS 缓存的 aiohttp 会话
• 使用 FastAPI lifespan 在启动时创建连接池，并在关闭时关闭
• 使用 Prometheus 指标监控连接池

连接池大小公式

pool_size = (concurrent_requests / avg_queries_per_request) * 1.5

该公式用于计算单个进程的连接池大小。针对服务端 max_connections 的集群级上限以及连接池告警阈值，参见 references/ork-delta.md。

反模式（禁止）

# NEVER use gather() for new code - no structured concurrency
# NEVER swallow CancelledError - breaks TaskGroup and timeout
# NEVER block the event loop with sync calls (time.sleep, requests.get)
# NEVER use global mutable state for db sessions
# NEVER skip dependency injection (create sessions in routes)
# NEVER share AsyncSession across tasks (race condition)
# NEVER use sync Session in async code (blocks event loop)
# NEVER create engine/pool per request
# NEVER forget to close pools on shutdown

上游覆盖范围（不重述）

在 2026-07-31 的 wrap-plus-delta 精简中移除的主题。请查阅一手来源；只有 ork delta（内部策略、经验教训、工作配置）属于此技能。如果某一行注明内部子集保留在 rules/ 文件中，该文件仍是 OrchestKit 立场的权威来源，而链接仅涵盖其周边的供应商内容。

主题	一手资料
asyncio 任务 API 参考：TaskGroup 与 gather()、asyncio.timeout()、ExceptionGroup 和 except*、to_thread()。项目内的子集保留在 rules/asyncio-taskgroup.md 和 rules/asyncio-cancellation.md 中。	https://docs.python.org/3/library/asyncio-task.html
asyncio.Semaphore、Lock、Event、Queue 的语义。项目内的子集保留在 rules/asyncio-structured.md 中；仅创建一次并加上超时的规则见 references/ork-delta.md。	https://docs.python.org/3/library/asyncio-sync.html
事件循环调试模式、slow_callback_duration、检测阻塞调用（项目内的 100 ms 阈值见 references/ork-delta.md）	https://docs.python.org/3/library/asyncio-dev.html
FastAPI 项目脚手架：应用布局、APIRouter 组合、Pydantic Settings 接线、uvicorn 入口点	https://fastapi.tiangolo.com/tutorial/bigger-applications/
FastAPI lifespan API 和启动／关闭机制。项目内的子集保留在 rules/fastapi-background.md 中；逆序清理规则见 references/ork-delta.md。	https://fastapi.tiangolo.com/advanced/events/
Starlette/FastAPI 中间件 API、BaseHTTPMiddleware、call_next、CORS 选项。项目内的子集保留在 rules/fastapi-middleware.md 和 references/fastapi-app-boilerplate.md 中；中间件与依赖项的划分见 references/ork-delta.md。	https://fastapi.tiangolo.com/tutorial/middleware/
SQLAlchemy 2.0 asyncio API：create_async_engine、async_sessionmaker、expire_on_commit、Mapped/mapped_column、with_for_update、批量插入和更新。项目内的子集保留在 rules/sqlalchemy-sessions.md、rules/sqlalchemy-relationships.md、rules/sqlalchemy-migrations.md 和 references/eager-loading.md 中。	https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
SQLAlchemy 连接池实现和选项（pool_size、max_overflow、pool_pre_ping、pool_recycle、pool_timeout）。项目内的子集保留在 rules/pooling-database.md 中。	https://docs.sqlalchemy.org/en/20/core/pooling.html
用于检测的 SQLAlchemy 连接池事件（checkout、checkin、connect）。项目内的子集保留在 rules/pooling-tuning.md 中；告警阈值见 references/ork-delta.md。	https://docs.sqlalchemy.org/en/20/core/events.html
asyncpg 连接池 API：create_pool、min_size/max_size、max_inactive_connection_lifetime、setup 钩子、类型编解码器	https://magicstack.github.io/asyncpg/current/api/index.html
aiohttp 客户端参考：ClientSession、TCPConnector 限制、keep-alive、DNS 缓存、ClientTimeout。项目内的子集保留在 rules/pooling-http.md 中。	https://docs.aiohttp.org/en/stable/client_reference.html
PostgreSQL 服务端连接限制（max_connections、superuser_reserved_connections）	https://www.postgresql.org/docs/current/runtime-config-connection.html
pytest-asyncio fixtures 和异步测试设置	https://pytest-asyncio.readthedocs.io/en/stable/reference/fixtures/index.html
容器和 Kubernetes 打包、就绪／存活探针、资源限制	https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
身份验证流程、密码哈希、令牌过期策略、安全标头	由 ork:security-patterns（src/skills/security-patterns/）负责；FastAPI Depends() 身份验证链保留在 rules/fastapi-dependencies.md 中
RFC 9457 问题详情正文格式、API 版本控制、SSE 和 WebSocket 线路契约	由 ork:api-design（src/skills/api-design/）负责；FastAPI 异常处理程序接线保留在 rules/fastapi-background.md 和 references/fastapi-app-boilerplate.md 中
FastAPI、asyncio、SQLAlchemy 和连接池的生产环境检查清单	可根据上述资料整理；不保留重复的检查清单

相关技能

• ork:architecture-patterns - 整洁架构与分层
• ork:async-jobs - 使用 Celery/ARQ 进行后台处理
• ork:api-design - 接口契约、RFC 9457 错误、SSE/WebSocket 流式传输
• ork:database-patterns - 数据库模式设计
• ork:security-patterns - 身份验证、密码哈希、令牌策略
• ork:testing-integration - pytest-asyncio 和 httpx ASGI 测试配置
• ork:devops-deployment - Docker 和 Kubernetes 打包