跳转至

FastAPI

FastAPI、Django、Flask

  • 重型企业级全栈项目，选Django(自带admin后台)
  • 纯API服务且追求极致性能，选FastAPI
  • 轻量灵活、需要高度定制或做胶水层，选Flask

核心架构与性能

对比维度 Django FastAPI Flask
框架定位 全栈重量级（内置一切） API专用轻量级（极致性能） 微内核核心（高度自由）
接口协议 WSGI（原生）
ASGI（实验性，支持不佳）		 ASGI（底层Starlette，原生异步） WSGI（原生）
async仅为语法糖，底层不变
并发模型 同步阻塞（多进程/多线程） 事件循环（单线程异步非阻塞） 同步阻塞（多线程，受限于GIL）
基准性能 中等（较重，启动慢） 最高（I/O密集场景领先2~3倍） 中等（轻量启动快，但并发能力弱）
WSGI/ASGI
  • WSGI: 标准的通信协议，同步阻塞；适合CPU密集型业务，但高并发下线程开销极大。
  • ASGI: 异步非阻塞 + 事件循环，使单个线程可撑起大量并发；使Python拥有Node.js/Go那样的高并发能力，适合IO密集型（调数据库、调第三方API）的微服务。
  • 语法糖: 让代码写起来更爽、读起来更优雅的简化写法，没有改变底层逻辑。
  • 胶水层: 适配转换头，用来做协议转换、数据清洗、路由转发等。

开发体验与生态

对比维度 Django FastAPI Flask
开箱即用程度 极高（内置Auth、表单、中间件） 中等（自带校验和文档，缺ORM/鉴权） 极低（仅路由和模板，其余全手搭）
API文档生成 需集成drf-yasg等第三方库 自动生成Swagger/OpenAPI（零配置） 需集成Flask-RESTX等第三方库
ORM与数据库 强绑定自带ORM（自动迁移，跨库） 无绑定（常配SQLAlchemy异步/SQLModel） 无绑定（社区标配Flask-SQLAlchemy）
灵活性（自由度） 极低（Django生态强约束） 较高（组件可自由组装） 最高（每个组件都可随意替换）
实现同等API代码量 最少 适中 最多（需手动处理序列化/校验）

适用场景与选型决策

应用场景 首选框架 核心理由（面试话术）
复杂ERP、电商主站、OA系统 Django Admin后台是杀手锏，内置功能避免造轮子，长期维护成本最低。
高并发对外API、微服务网关、ML模型部署 FastAPI 异步高性能 + 自动生成API文档 + 类型安全（Pydantic），前后端联调极快。
轻量级单体服务、快速原型验证、物联网网关 Flask 资源占用极低，启动速度最快，且侵入性最小，适合做胶水层。
遗留系统改造或中间件（BFF）层 Flask 代码透明，极易与老旧同步代码集成，改造成本最低。

高阶陷阱与认知深度

高阶知识点 Django / Flask（同步阵营） FastAPI（异步阵营）
异步的真实性 异步支持基本是语法糖，底层Worker仍是同步阻塞，无法突破GIL。 原生异步，但必须搭配asyncpg或异步ORM才能发挥威力。
常见的性能陷阱 并发量受限于线程池大小，调高线程数会导致上下文切换开销剧增。 若误用了同步阻塞的SQLAlchemy或未配置数据库连接池，性能会直接退化到Flask水平。

SQLAlchemy 2.0 支持同步和异步两种方式: psycopg2(同步阻塞)、asyncpg(异步非阻塞)。

工程能力

数据库连接池

  1. 连接池显示参数: 最小/最大值、客户端查询超时时间、服务端兜底超时时间（杀掉慢SQL）
  2. 慢SQL治理与监控: 除了索引，需检查事务中的外部IO，如生成任务（提取测试点 → 生成 → 完善 → 格式化，每步都调 LLM API，总耗时可达数分钟）整体包在 in_transaction("postgres") 里。这意味着每个生成任务独占 1 个池连接长达几分钟。
  3. 看板数据缓存: 非必须实时的数据，考虑进程内 TTL 缓存（1~5 分钟）或 Redis 缓存

异步任务与事件循环

Q: FastAPI 的 async/await 在底层是如何通过 asyncio 事件循环调度的？如果我在路由函数里写了一个同步的 CPU 密集型计算（比如大循环），会如何影响其他并发请求？你是如何规避的？”

A: async def 路由运行在主事件循环；若直接调同步CPU密集型计算，会阻塞整个事件循环，导致该 Worker 无法处理任何其他请求，QPS 会断崖式下跌。

解决方案

  • 轻度计算: 在 async 路由中，使用 loop.run_in_executor(None, sync_func) 将其委托给默认的 ThreadPoolExecutor 执行，避免阻塞。
  • 重度计算（如加密/大数据聚合）: 则使用 ProcessPoolExecutor 或直接解耦为独立"待处理任务表 + Celery"定时任务。

追问: 基于"待处理任务表 + Celery"定时任务，若连接池满，新进来的请求只能阻塞等待直到超时。单靠 Python 代码，如何设计一个简易的‘熔断’或‘快速失败’逻辑？

A: 从接口层的快速拒绝

  • 排队阈值熔断: 接口收到请求时，先 COUNT 任务表中 status='PENDING' 的数量。如果超过设定阈值（比如 1000 条），直接返回 HTTP 429（Too Many Requests），不再往 DB 插入新任务。这本质上就是利用了 DB 作为共享计数器，实现分布式限流。
  • 超时快速失败: 在 SQLAlchemy 层面设置 pool_timeout=5s，一旦获取连接超时，立刻抛出异常并返回 503，防止请求线程无限阻塞，耗尽 Tomcat/UVicorn 的工作线程。

(Agent)任务执行时服务重启，如何恢复，同时避免从头执行

核心: 恢复重跑，先查上次执行的准确结果，从断点继续跑；跳过已成功的节点，避免从头重复执行。

  1. 唯一任务标识，重跑时判断上次任务执行是否成功，尤其外部工具重复调用敏感时，应具备幂等性。
  2. checkpoint持久化至数据库，将agent任务关键节点切片化，恢复重跑或结果不满意部分重跑从指定节点继续执行。内容包括: 当前节点、阶段输入与产物、工具调用结果、后续步骤状态等。
  3. 状态管理机制，节点有明确的状态划分如: 待执行、执行中、执行完成、执行失败。