FastAPI 入门：一个普通 Python 后端工程师的真实踩坑记
大家好，我是阿凯。今天我想和你们聊聊我第一次用 FastAPI 开发后端项目的经历。
说实话，刚开始接到这个项目时，我心里是有点打鼓的。我们团队之前一直是用 Flask 做服务端开发，虽然稳定、熟悉，但面对越来越复杂的接口需求和性能瓶颈，也开始觉得有些力不从心了。尤其是当我们要开始做一个用户行为日志收集系统的时候，传统的同步框架明显不够看了。
所以，我们决定尝试一下社区里讨论得越来越多的一个新玩意儿——FastAPI。
初识 FastAPI：不只是另一个“微框架”
先说下我们的项目背景吧：
我们要做的是一个用户行为埋点收集与分析平台，主要功能包括：
- 接收客户端发送的埋点数据（GET/POST）
- 对埋点做基础校验（字段是否完整、格式是否正确等）
- 存储到 ClickHouse
- 提供简单查询接口给 BI 系统
这个系统要支撑百万级 PV 的流量，同时尽可能降低延迟。
一开始我觉得：“嗯，还是继续用 Flask + Celery 异步处理吧”。但是当我翻文档看到 FastAPI 支持异步请求、自动生成 OpenAPI 文档，并且有 Pydantic 做数据模型验证之后，我就改主意了——“这不就是我们需要的轻量又灵活的新一代框架吗？”
而且更重要的是，它居然能和异步数据库驱动配合使用，比如 asyncpg 和 motor（MongoDB 异步驱动），这对提升整体吞吐非常关键。
遇到的第一个问题：接口定义怎么写才规范？
刚开始上手 FastAPI 的时候最让我困惑的，不是它有多快或多强大，而是“我该怎么写接口才算合理？”
以前在 Flask 里面写接口，基本就是 route 加 view 函数，参数随便取一取。但在 FastAPI 中，你需要用 Pydantic 模型 来明确定义请求体结构和响应体格式。
举个例子，我们要接收一个埋点数据：
from pydantic import BaseModel
class TrackingEvent(BaseModel):
user_id: str
event_type: str
properties: dict
timestamp: float
然后你的 POST 接口就能自动解析请求中的 JSON 并进行类型校验：
@app.post("/events")
async def receive_event(event: TrackingEvent):
# 做一些处理...
return {"status": "ok"}
这么写完以后，不仅可以通过 /docs 自动生成 Swagger UI，还可以减少很多手动校验逻辑，甚至可以统一错误返回结构。这种体验真的很爽，特别是当你需要频繁调试不同环境传参的时候。
不过刚换过来那几天，我也犯过不少低级错误，比如忘记写 BaseModel 或者字段名拼错了，结果 FastAPI 就报错让你改。起初我很抗拒，总觉得限制太多，后来才发现这是它帮我们提前发现了潜在的问题。
数据库连接慢？试试 async 驱动！
接下来，真正让我感受到 FastAPI 性能优势的地方，是和数据库打交道这一块。
我们最初是用普通的 SQLite+SQLAlchemy 同步方式来做原型开发，速度还可以接受。但一旦接入真实数据流（每秒几千条埋点），就开始出现大量阻塞。
于是我做了第一轮改造：换 ClickHouse + 使用异步写入驱动。
我们选择了一个第三方的异步 ClickHouse 客户端（clickhouse-async），它支持通过 asyncio 发起非阻塞插入操作。
代码大致长这样：
from clickhouse_async import dbapi
@app.post("/events")
async def receive_event(event: TrackingEvent):
conn = await dbapi.connect(host='localhost', port=8123, database='tracking')
cursor = await conn.cursor()
await cursor.execute(
"INSERT INTO events (user_id, event_type, properties, ts) VALUES",
[(event.user_id, event.event_type, event.properties, event.timestamp)]
)
return {"status": "ok"}
别看这段代码简单，实际跑起来之后，系统的吞吐能力提升了至少 3~4 倍。特别是在并发请求较多的情况下，FastAPI + async DB 配合的效果特别明显。
不过这里我也踩了个大坑 —— 没有复用数据库连接池！
起初为了图方便，每次写入都重新建立连接，结果随着并发增加，系统就卡死了。后来改成用单例模式维护了一个连接池，才缓解了压力。这也提醒我们，即使是异步代码，也要注意资源管理。
生产上线之前的优化：中间件、缓存、日志
在准备上生产的过程中，我还做了一些额外的优化，想分享给大家：
1. 添加中间件记录请求耗时
我们加了一个简单的中间件，用来统计每个请求的耗时情况，用于监控性能波动：
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
logger.info(f"{request.url} processed in {process_time:.4f}s")
return response
这玩意儿虽小，但对定位性能问题真的很有用。
2. 缓存常用配置项
我们在某些高频读取场景中引入了 Redis 缓存，避免重复查库。比如埋点事件白名单、用户权限等。
这部分用了 redis-py 的异步版本 aredis，效果挺不错：
import aredis
redis_client = aredis.StrictRedis(host="redis-host", port=6379)
@app.get("/feature-flag/{flag_name}")
async def get_flag(flag_name: str):
flag_value = await redis_client.get(f"flag:{flag_name}")
return {"flag": flag_name, "value": flag_value}
当然你也可以结合 TTL 设定缓存时间，避免缓存穿透或者雪崩。
3. 日志和健康检查接口
FastAPI 默认的日志信息其实不太全，所以我们自己封装了一套基于 uvicorn 的 logging handler，并加上了 trace ID、调用栈等信息，方便后续排查问题。
另外我们也加了健康检查接口：
@app.get("/healthz")
async def health_check():
return {"status": "healthy"}
部署到 Kubernetes 以后，这个接口被用作 readiness/liveness probe，帮助我们更好地掌握服务状态。
上线后的挑战：性能优化和运维监控
项目上线一段时间后，也遇到了一些新的问题。
首先是并发压力测试没过：压测工具模拟 500QPS 的时候，发现接口有明显的延迟毛刺。我们用了几个手段来解决：
✅ 增加 Worker 数量
FastAPI 是基于 ASGI 的框架，不像 Flask 那样只能用多线程或 Gunicorn 多进程。我们直接用 Uvicorn + 多 worker 启动，提升并发能力。
启动命令如下：
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
注意：如果你使用 Uvicorn 启动多个 worker，请确保数据库连接池、Redis 连接等资源是共享或合理分配的。
✅ 使用 Prometheus + Grafana 做监控
为了更细致地观察服务运行状况，我们在项目中引入了 Prometheus 的 middleware：
from starlette.middleware import Middleware
from starlette.middleware.prometheus import PrometheusMiddleware, metrics
middleware = [
Middleware(PrometheusMiddleware),
Middleware(SessionMiddleware, secret_key="some-secret-key"),
]
app = FastAPI(middleware=middleware)
app.add_route("/metrics", metrics)
然后搭配 Grafana 把 API 请求延迟、成功率、并发数等指标可视化，大大提升了我们的可观测性。
经验总结：FastAPI 的优缺点及适用场景
经过这次实战，我对 FastAPI 有了更深入的认识，也总结了几点经验，希望对你有所帮助：
✔️ 适合使用的场景
- 需要高性能、高并发的 API 服务（如埋点收集、实时推荐等）
- 团队对 Python 异步生态有一定了解，愿意学习 aiohttp / asyncpg / motor 等异步库
- 重视接口文档一致性，追求自动化测试与文档维护效率
❌ 不太推荐的场景
- 只是写个内部脚本或者小型管理后台
- 团队对异步编程无基础，也不愿意投入时间学习
- 现有项目已经用了成熟稳定的 Flask/Django 架构，无需升级
💡 个人建议
- Pydantic 真香预警：不要怕麻烦，老老实实写数据模型，接口健壮性会大幅提升。
- 数据库选型很重要：如果用同步驱动，FastAPI 的异步优势就浪费了，一定要根据业务特性合理选型。
- 中间件和日志别省事：越早加上越好，不然排查线上问题会很痛苦。
- 文档即接口规范：Swagger UI 自动生成的文档一定要同步更新，对前后端协作非常重要。
写在最后：技术只是工具，解决问题才是目标
这篇文章讲了很多技术细节，但我希望大家记住的核心不是某个函数怎么写，也不是哪个包最好用。
而是：
在面对复杂业务场景时，如何选择合适的技术方案，并一步步把它落地、优化、最终交付上线，是一个工程师真正的价值所在。
FastAPI 是一个很好用的工具，但它只是一个起点。在项目推进过程中，你会发现性能优化、日志监控、链路追踪、灰度发布等等环节，才是真正考验你工程能力的部分。
就像我在埋点系统上线前那两周，凌晨三点还在修 Redis 连接泄漏 bug 一样，技术路上总是充满了各种不确定性和挑战。
但正是这些“折腾”，让我们不断成长。
如果你正在考虑入门 FastAPI，或者已经在用它但还处在迷茫期，希望我的这段亲身经历能够带给你一些启发。
欢迎留言交流你自己的项目经验，我们一起进步！
附录：推荐阅读资料
祝你在 FastAPI 的旅程中越走越远～
评论 0