同步引擎用于 Alembic 迁移
FastAPI 入门：Python 后端开发新手指南
开篇：为什么我选择用 FastAPI 来做后端？
作为一个常年混迹于 Python 领域的开发者，从最初的 Flask 到后来的 Django，再到如今的 FastAPI，我对各种 Web 框架都有过实战体验。但真正让我坚定地把重心转向 FastAPI 的，是一次项目重构的经历。
那是去年的一个中型在线教育平台项目，我们原本基于 Django 构建了一个 API 系统，随着业务增长，接口响应速度、文档维护、类型安全等问题越来越突出。尤其是前端团队对接时经常因为字段不明确、参数格式错误导致调试耗时巨大。更糟糕的是，性能在高峰期也频频出现瓶颈。
当时我在技术选型时接触到了 FastAPI，一开始其实心里有些疑虑：一个“新”框架能靠谱吗？结果一试之下彻底改观——它不仅拥有接近 Go 语言级别的性能表现（得益于异步特性），还自带交互式文档（Swagger 和 ReDoc）、支持 OpenAPI 标准、代码简洁优雅，并且最关键的：类型系统开箱即用，几乎零学习成本就能写出可读性很高的 API 接口。
所以今天我想通过自己亲身经历，带着你快速上手 FastAPI，并把它落地到实际项目中。如果你是个刚入行或者想转 Python 后端的新手，这篇文章可能会成为你成长路上的一盏灯；如果你已经有一定经验，希望也能从中发现一些新思路和踩坑经验。
项目背景：教育平台用户中心重构
我们的项目目标是为一个在线教育平台重构用户中心模块，包括用户注册、登录、信息管理、课程记录、消息通知等核心功能。原来的 Django 系统在数据量增加之后变得迟缓，特别是在处理并发请求时，经常出现连接池打满的情况。
我们需要一个新的后端架构：
- 更高的吞吐能力
- 支持异步操作（比如邮件发送、第三方认证）
- 快速生成文档并支持自动测试
- 数据库访问优化
- 类型安全以减少接口错误
FastAPI 几乎完美符合所有需求。
问题描述：Django 给我们带来的痛点
虽然 Django 是一个非常成熟的框架，但在高并发场景下，它的阻塞性质和 ORM 的延迟加载设计，在没有深度优化的情况下很容易成为瓶颈。
具体问题包括：
- API 文档维护困难：虽然有 Django REST framework 提供了文档支持，但需要手动写很多注释，而且不够直观。
- 异步处理不便：Django 对异步的支持直到 3.1 才开始逐渐完善，而我们在使用 Celery 做任务队列的时候总是感觉“别扭”，尤其是在调用数据库的时候容易遇到线程安全问题。
- 接口参数类型不安全：很多字段传递进来的时候没有强类型检查，导致调试时间变长。
- 数据库性能下降：在并发请求较高时，PostgreSQL 连接池经常被打满，查询慢、事务锁等问题频发。
这些问题让我们意识到，是时候换条赛道了。
解决方案：为什么是 FastAPI？
FastAPI 之所以成为我们的首选，主要原因如下：
- 基于 Starlette 构建，天生支持异步编程
- 自动生成 Swagger UI 和 ReDoc 接口文档
- 与 Pydantic 集成，天然支持数据模型验证
- 轻量、灵活，便于扩展微服务结构
- 社区活跃，生态成熟
更重要的是，它可以很自然地配合 SQLAlchemy 或 Tortoise ORM 使用，并且可以通过 asyncpg 或 aiomysql 实现异步数据库操作，这对于提升整个系统的吞吐量至关重要。
代码实践：从零搭建一个 FastAPI 服务
下面我会带你一步步搭建一个 FastAPI 服务，并接入 PostgreSQL 数据库。我们将构建一个简单的用户注册接口作为示例。
1. 安装依赖
pip install fastapi uvicorn sqlalchemy pydantic asyncpg
2. 初始化项目结构
project/
├── main.py # FastAPI 主文件
├── models.py # ORM 模型定义
├── schemas.py # 接口输入输出定义
├── database.py # 数据库连接配置
└── routes.py # 路由定义
3. 数据库配置：database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
engine = create_engine("postgresql://user:password@localhost/mydb")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 异步引擎用于运行时
async_engine = create_async_engine(
"postgresql+asyncpg://user:password@localhost/mydb",
pool_size=10,
max_overflow=20
)
AsyncSessionLocal = sessionmaker(
class_=AsyncSession,
bind=async_engine,
expire_on_commit=False
)
Base = declarative_base()
4. 用户模型：models.py
from sqlalchemy import Column, String, Integer
from database import Base
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True)
email = Column(String(100), unique=True)
hashed_password = Column(String(100))
5. 输入/输出模型定义：schemas.py
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
email: str
password: str
class UserOut(BaseModel):
id: int
username: str
email: str
class Config:
orm_mode = True
6. 创建路由：routes.py
from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session
from typing import List
from database import get_db
from models import User
from schemas import UserCreate, UserOut
router = APIRouter()
@router.post("/users/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
new_user = User(**user.dict())
db.add(new_user)
db.commit()
db.refresh(new_user)
return new_user
@router.get("/users/", response_model=List[UserOut])
async def read_users(db: Session = Depends(get_db)):
users = db.query(User).all()
return users
注意：以上为同步 ORM 写法。如果你要使用异步 ORM，可以替换
get_db为异步函数，并用await db.execute()等方式调用。
7. 主应用启动文件：main.py
from fastapi import FastAPI
from routes import router
import uvicorn
app = FastAPI(title="FastAPI Demo", version="1.0")
@app.on_event("startup")
async def startup_event():
print("Service is starting up...")
@app.on_event("shutdown")
async def shutdown_event():
print("Shutting down...")
app.include_router(router, prefix="/api")
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
现在你可以启动服务：
uvicorn main:app --reload
访问 http://localhost:8000/docs，你会看到自动生成的 Swagger UI 文档页面，点击任意接口可以在线测试。
踩坑经验：那些让我深夜失眠的 Bug
虽然 FastAPI 性能好、易上手，但在实际开发过程中我也遇到了不少“坑”，以下是我整理的一些常见问题及解决方案：
1. 数据库连接池打满的问题
初期我们使用了默认的连接池设置，在并发量稍微上去之后就频繁报错：
psycopg2.pool.TooManyConnections
解决方案很简单，在数据库引擎初始化时加入连接池设置：
async_engine = create_async_engine(
"postgresql+asyncpg://user:password@localhost/mydb",
pool_size=10,
max_overflow=20
)
这相当于设置了最大连接数不超过 30。
2. 异步 ORM 使用不当导致性能倒退
在迁移到异步版本的过程中，我们一度误用了 db.session.add 等同步方法，结果反而比之前的同步版本还要慢。
正确的做法是使用异步 ORM：
from sqlalchemy.ext.asyncio import AsyncSession
async def some_async_method(db: AsyncSession):
result = await db.execute(select(User).where(...))
user = result.scalars().first()
确保在整个链路中保持异步调用才能发挥最大性能。
3. Pydantic 模型无法识别 ORM 模型
当你尝试直接将 ORM 模型返回给客户端时会收到类似错误：
TypeError: Object of type User is not JSON serializable
解决办法是在 schema 的 Config 中声明 orm_mode = True：
class UserOut(BaseModel):
id: int
username: str
class Config:
orm_mode = True
4. Swagger 页面空白，文档无法加载
有时候部署到生产环境后，Swagger 页面显示一片空白。原因是静态资源路径没配置好。解决方法是在主程序中加入以下代码：
app.mount("/static", StaticFiles(directory="static"), name="static")
app.openapi_url = "/api/openapi.json"
app.docs_url = "/docs"
app.redoc_url = None
确保你的静态目录中有相关资源，否则可能需要重新安装 Starlette 相关包。
效果总结：重构后的收获
项目上线半年后，我们做了效果复盘：
| 指标 | 重构前 | 重构后 |
|---|---|---|
| 平均接口响应时间 | 280ms | 95ms |
| QPS | ~300 | ~1200 |
| 文档更新效率 | 手动编写耗时 | 自动生成，即时可用 |
| 前后端协作效率 | 易出错 | 接口定义清晰，沟通成本降低 |
此外，由于有了更强的类型校验，接口调用的异常明显减少，运维日志中“非法参数”的错误减少了 90% 以上。
最让人开心的是，新的代码结构让新成员上手更快，接口逻辑更加清晰。
经验分享：给 Python 新手的建议
如果你是一个刚刚踏入后端领域的开发者，或者正准备从 Flask / Django 转向 FastAPI，我建议你：
✅ 学会 Pydantic，它是 FastAPI 的灵魂
Pydantic 提供了强大但简单易懂的数据模型验证能力。你甚至可以用它来做后台逻辑校验、数据转换、嵌套结构解析等工作。
class RegisterRequest(BaseModel):
email: EmailStr
password: constr(min_length=6)
confirm_password: str
@validator('confirm_password')
def passwords_match(cls, v, values, **kwargs):
if 'password' in values and v != values['password']:
raise ValueError('两次密码不一致')
return v
这样的代码不仅能自动解析 JSON 参数，还能在出错时给出精确提示，大大提升了接口健壮性。
✅ 利用异步优势，但不要滥用
不是每个接口都值得异步化。比如像用户注册、支付回调这种业务逻辑比较重的接口，适合异步执行。而像列表查询、详情页这种，同步处理反而更高效。
✅ 设计数据库时多考虑索引和关系
很多人忽视了数据库设计的重要性。即使再快的接口，如果数据库设计不合理，也会拖垮整个系统。
常见的建议有：
- 给外键、常用查询字段添加索引
- 避免大表联查，提前做好分表设计
- 尽量避免 N+1 查询问题，使用 JOIN 加载关联数据
✅ 上线之前务必做压力测试
工具推荐 Locust，编写测试脚本非常简单：
from locust import HttpUser, task
class MyUser(HttpUser):
@task
def register(self):
payload = {
"email": "test@example.com",
"password": "123456",
"confirm_password": "123456"
}
self.client.post("/register", json=payload)
运行 locust 之后模拟数千并发请求，观察接口响应时间和错误率是否正常。
最后的小插曲：关于技术选型的思考
有一次，我们在内部技术评审会上讨论是否应该继续使用 FastAPI。一位老工程师提出了质疑：“FastAPI 太新了，万一将来不再维护怎么办？” 这个问题其实我也曾思考过。
我的回答是：“FastAPI 的作者 Sebastian Ramirez 也在积极参与 FastAPI 社区，GitHub 星星已经超过 40k，背后有 Tiangolo 等开源组织在支撑，我相信它未来几年都不会消失。”
果然，今年 Python 核心开发者 Guido van Rossum 也公开表示他喜欢 FastAPI，这让很多人都对这个框架更有信心了。
结语：FastAPI 让我重新爱上写后端
回顾这一路的转型之路，从最初对 FastAPI 的怀疑，到后来完全依赖它构建高性能服务，我深刻体会到一个好的工具真的能让工作变得更轻松、更有乐趣。
如果你也在寻找一门现代、高效、易于维护的 Python Web 框架，不妨试试 FastAPI。
也许下一个爆款项目，就是你写的。
评论 0