从零开始搭建一个Python后端服务：FastAPI入门实践手记
去年我刚加入一家初创公司，负责搭建内部的API网关和一些基础业务模块。团队技术栈主要是Python，之前用的是Flask做后端框架，但随着项目越来越复杂，接口数量增加、响应速度要求变高，传统的Flask应用逐渐显得吃力，尤其是在处理异步请求和自动文档生成方面，总是需要手动维护很多东西，效率很低。
那时候我就在想，有没有更好的方式来写Python后端？能不能找一个既轻量又性能好，还能支持现代Web开发特性的框架？于是，FastAPI进入了我的视野。
这篇文章，我想以我在实际项目中初次上手FastAPI的经历为主线，讲讲我是怎么一步步从头构建一个简单却实用的后端服务的，过程中遇到的问题、踩过的坑，以及最终上线后的效果。
背景与项目需求：我们为什么需要这个服务？
我们的项目是一个企业内部使用的用户行为分析平台，其中有一部分是收集前端埋点事件数据，进行初步清洗和存储，供后续数据分析使用。最开始这个功能是在主服务中直接处理，但随着埋点量激增，导致主服务压力山大，接口响应慢甚至超时的情况越来越多。
所以，我们决定把这个埋点接收和处理模块拆出来做成独立的服务。目标很简单：
- 提供一个高性能的HTTP接口，用于接收大量并发的小型POST请求（每个请求大约1~2KB）
- 数据要能快速写入数据库，并且尽量不影响主线业务
- 需要有清晰的文档和接口说明，方便其他团队接入
- 开发周期要尽可能短，因为我们急需上线缓解当前压力
技术选型：为什么选择 FastAPI？
当时团队里对Python后端的了解集中在两个框架：Flask 和 Tornado。Flask很熟悉，但我们已经意识到它在处理并发上有瓶颈；Tornado虽然原生支持异步，但它的语法风格比较独特，学习成本略高，而且社区生态没有那么活跃。
而这时，一位同事向我推荐了 FastAPI，说是目前 Python 社区里增长最快的现代 Web 框架之一。我花了一天时间研究文档后发现，这简直是为我们这个项目量身定做的：
FastAPI 的核心优势：
- 基于 Starlette + Pydantic，天生支持异步请求处理；
- 自动生成 OpenAPI 文档和 Swagger UI，接口调试非常方便；
- 类似 Flask 的极简风格，但性能更好，代码结构也更清晰；
- 内置参数校验系统（Pydantic），避免了很多重复的字段检查工作；
- 社区活跃，文档丰富，适合团队协作。
最重要的一点是，它可以让我在很短时间内完成开发、测试，并迅速部署到生产环境。
实践过程：从搭建到上线的全过程
第一步：初始化项目结构
为了更好地组织代码，我一开始就把项目结构做了个简单设计，基本如下：
event-collector/
├── app/
│ ├── main.py
│ ├── models.py
│ ├── schemas.py
│ ├── database.py
│ └── routers/
│ └── events.py
├── requirements.txt
└── config.yaml
main.py是 FastAPI 的主入口models.py定义 SQLAlchemy 的 ORM 模型schemas.py使用 Pydantic 定义请求体和返回格式database.py负责数据库连接池配置routers/下放置各个模块的接口路由config.yaml存放配置信息
这样的结构便于后期维护，尤其当你服务越来越复杂的时候尤为重要。
第二步：编写第一个接口
我们的第一个需求是：接收客户端发送的埋点数据，并存入数据库。
定义一下 POST 请求的 Body 结构：
from pydantic import BaseModel
from datetime import datetime
class EventCreate(BaseModel):
event_type: str
user_id: str
payload: dict
timestamp: datetime = None
然后，我们在 app/routers/events.py 中定义一个简单的接口：
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Event
from app.schemas import EventCreate
router = APIRunner(prefix="/events", tags=["events"])
@router.post("/")
def create_event(event: EventCreate, db: Session = Depends(get_db)):
db_event = Event(**event.dict())
db.add(db_event)
db.commit()
db.refresh(db_event)
return {"status": "success", "id": db_event.id}
是不是看起来特别像 Flask？但区别在于，这里我们不需要再手动解析 JSON、也不需要手动验证参数类型——一切都由 FastAPI 自动完成，出错会直接返回 422 错误。
第三步：数据库操作
我们使用 PostgreSQL 作为数据库，ORM 框架选择 SQLAlchemy。
首先在 database.py 中配置连接：
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://user:password@localhost/event_db")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
接着在 models.py 定义模型：
from sqlalchemy import Column, String, JSON, DateTime, Integer
from app.database import Base
from datetime import datetime
class Event(Base):
__tablename__ = 'events'
id = Column(Integer, primary_key=True, index=True)
event_type = Column(String(100), index=True)
user_id = Column(String(100))
payload = Column(JSON)
timestamp = Column(DateTime)
FastAPI 本身不强制使用某种 ORM，但它可以很好地和 SQLAlchemy 或者 Tortoise ORM 等一起使用，这点非常灵活。
第四步：异步写入优化性能
虽然上面的接口已经能满足基本需求，但在高并发下性能并不理想。因为默认情况下 SQLAlchemy 的 session 是同步的，每次请求都需要等待数据库 IO 完成。
于是我们考虑使用异步模式，借助 Starlette 原生支持的 async 特性：
改造数据库连接为异步
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
ASYNC_DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
engine = create_async_engine(ASYNC_DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_async_db():
async with AsyncSessionLocal() as db:
yield db
注意要把依赖改成 async：
@router.post("/")
async def create_event(event: EventCreate, db: AsyncSession = Depends(get_async_db)):
db_event = Event(**event.dict())
db.add(db_event)
await db.commit()
await db.refresh(db_event)
return {"status": "success", "id": db_event.id}
这样就可以利用异步IO的优势，提高整体吞吐量。
第五步：日志与监控
服务上线前，我给它加了些日志和健康检查接口。
FastAPI 内置了 /docs 和 /redoc 接口文档页，非常方便；同时我还加了一个 /health 接口：
@router.get("/health")
def health_check():
return {"status": "OK"}
另外，使用 logging 模块记录关键操作：
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
logger.info("Received event from user %s", event.user_id)
上线后还接入了 Prometheus 监控，在中间件里暴露指标端点，监控 QPS、延迟、错误率等核心指标。
遇到的坑和解决办法
说实话，虽然是个小项目，但在真正部署到生产的时候，还是踩了一些坑。
1. Pydantic 模型字段缺失引发异常
刚开始我们有一个字段没设 default 值，结果前端有时候传不到，导致整个接口报错。后来通过 Pydantic 的 Optional 字段加上默认值解决了。
比如：
from typing import Optional
class EventCreate(BaseModel):
event_type: str
user_id: str
payload: dict
timestamp: Optional[datetime] = None
2. 异步数据库连接问题
异步环境下 SQLAlchemy 有些方法需要 await，不然会报错。比如 db.commit() 应该写成 await db.commit()，否则程序会卡死。
3. 部署时数据库迁移没做好
生产环境上线前忘记跑 migration，导致表不存在。教训：务必把数据库建模、migration 工具集成进 CI/CD 流程中。
我们后来改用 Alembic 做 migrations：
alembic init alembic
并在 CI 中自动执行升级脚本：
alembic upgrade head
上线效果与性能提升
我们用了 Uvicorn 启动服务，配置多个 worker 进行压测：
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
在本地测试环境下，QPS 达到了 2600+，比之前的 Flask 服务提升了近 5 倍。
在正式上线后，服务稳定性非常好，基本没出现过内存泄漏或响应迟缓的问题。配合 Nginx 做负载均衡之后，扛住了每天几百万条事件的写入压力。
总结与建议
从这次实践中，我对 FastAPI 的使用有了全面的认知。总结几个我学到的经验，分享给大家：
✅ 快速原型开发利器
- 如果你是从小规模项目起步，或者是需要快速搭出一个接口服务，FastAPI 几乎是最优选择。
- 它的自动生成文档功能，真的节省了非常多时间，特别是和其他团队联调的时候。
✅ 异步能力强大但需谨慎使用
- 异步接口确实能提升性能，但这需要你有一定的 asyncio 基础。
- 尽量确保数据库、网络请求等 IO 密集型任务也用异步方式处理，否则达不到预期效果。
✅ 架构设计要早做准备
- 即使是小项目，也要尽早考虑好模块划分、依赖管理、配置分离等问题。
- 用好分层架构，后期重构或者拓展功能会轻松很多。
✅ 不要忽视运维和监控
- 日志、健康检查、Prometheus 指标这些看似“附属”的功能，其实关系到服务能否稳定运行。
- 别等到出了线上事故才去补这些东西。
最后一点感悟
其实我一直觉得，一个好的后端开发者，不是只看他会写多少酷炫的算法，而是看他是否能在合适的时间选用合适的工具解决问题。
FastAPI 并不是什么万金油式的框架，但它确实帮我在一个紧张的时间内完成了需求，而且上线后的表现完全超出预期。
如果你还在用 Flask 或者 Django 写 RESTful API，真的不妨试试 FastAPI。也许你会发现，原来 Python 的后端开发也可以这么快、这么现代化。
如果你有任何关于 FastAPI 或者后端开发的疑问，欢迎留言交流。下一篇文章我会聊聊如何用 FastAPI + Celery 实现后台任务队列，敬请期待！
评论 0