SuperAPI AI 工具集

python-backend

编程与开发

查看项目

更新于 10-15 22:33

技能介绍

用途:该技能面向生产环境的 Python 异步后端开发,围绕 asyncio、FastAPI、SQLAlchemy 2.0 与连接池提供模式与规则,共 4 类 12 条规则,规则文件存放在 rules/ 目录下按需加载。

主要能力:Asyncio 类(3 条,高影响)覆盖 TaskGroup 结构化并发、取消处理与超时;FastAPI 类(3 条)覆盖依赖注入、中间件与后台任务;SQLAlchemy 类(3 条)覆盖异步会话、关系映射与迁移;Pooling 类(3 条)覆盖数据库池、HTTP 会话与调优。

必要用法(示例保持原样):

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

SessionDep = Annotated[AsyncSession, Depends(get_db)]

Asyncio TaskGroup with timeout

async with asyncio.timeout(30): async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(fetch_url(url)) for url in urls]

关键决策:使用 TaskGroup 而非 gather();用 asyncio.timeout() 管理超时;用 Semaphore 限制并发;用 asyncio.to_thread() 桥接同步代码;始终重新抛出 CancelledError;每请求一个 AsyncSession 且 expire_on_commit=False;关系设置 lazy="raise" 防止 N+1;集合用 selectinload 预加载;pool_pre_ping=True。池大小公式:pool_size = (concurrent_requests / avg_queries_per_request) * 1.5。

限制与禁止:不得用 gather() 写新代码,不得吞掉 CancelledError,不得用同步调用(time.sleep、requests.get)阻塞事件循环,不得使用全局可变状态保存 db 会话,不得跳过依赖注入,不得跨任务共享 AsyncSession,不得在异步代码中使用同步 Session,不得每请求创建引擎/池,不得忘记在关闭时释放池。部分话题(安全、错误体格式等)由其他技能负责,本技能不重述;本文未提供收费、价格及具体部署步骤等信息。