FastAPI 的一些感受
这不是教程。照着抄大概率跑不起来(表怎么建的我都没写),我就是想把这些天撞过的墙记下来。
前言
国庆之后,现实里一堆压力压过来,不得不边做项目边补自己的能力。想深入学 Python,就顺手往 Python 后端那边拐了,盯上了最流行的三个:Django、FastAPI、Flask(Django 就摸过一点皮毛,算我学浅)。在 FastAPI 和 Flask 之间我最后选了 FastAPI——理由很简单,它比 Flask 火。
现在回头看这理由挺扯的,不过结果还行。
吹响进攻的号角
启动
我电脑上装了 uv,跑起来就一行:
uv run uvicorn app:app --reload
app:app 是按 文件名:变量名 拼的,我正好两边都叫 app,看着跟写重了似的。--reload 是热重载,改完代码自己重启,只能开发的时候用。
起来之后打开 http://127.0.0.1:8000/docs,第一次看到那个能点、能直接发请求的页面——这半个多月最有成就感的两分钟就是那会儿。
路由
路由就长这样:
@app.get("/")
def hello_world():
return {"message": "Hello World"}
@ 是装饰器,app 是实例,get 换成 post、put、patch、delete 就是别的请求方法。get 是读,post 是建,put 整个换掉,patch 改一半(我项目里就用了 get/post/put/delete 四个)。
写上几个之后,文档就成这样了:
当时爽在哪:函数名、参数类型、返回值,你写啥它文档就是啥。我是到这儿才发现,原来“写代码”和“写文档”能是一件事。
pydantic
pydantic 是运行时校验类型的工具,最常用的是 BaseModel。
from pydantic import BaseModel
class UserBase(BaseModel):
name: str
age: int
请求体一进来先被它过一遍:类型不对、字段缺了,FastAPI 直接甩个 422 回去,压根走不到我函数里。
当时想通的一件事:以前校验全靠自己,一进函数就
if not isinstance(...)、if "name" not in data,一路 if 下去。现在这道墙挪到函数门口了——该信任谁这件事,从函数里面搬到了函数入口。我对“后端”这词第一次有点实感,就是这时候。
字段表还能拆成好几个模型换着用(我拆了 UserBase / UserCreate / UserRead / UserUpdate):UserCreate 带 password,UserRead 不带 hashed_password,所以响应里永远看不见密码;UserUpdate 全字段可选,专门拿来局部更新。
但我的返回值是 SQLAlchemy 的 User 对象,不是 dict,UserRead 一开始压根不认识它:
from pydantic import BaseModel, ConfigDict
class UserRead(UserBase):
id: int
model_config = ConfigDict(from_attributes=True) # 允许从 ORM 对象属性取值
这行我盯了半天才反应过来:response_model 能直接 return db_user,全靠它。
依赖注入
一开始我是真不懂「依赖注入」四个字啥意思,后来发现它就藏在参数里。拿 get_user 举个例子:
@app.get("/{user_id}", response_model=UserRead)
def get_user(user_id: int, db: Annotated[Session, Depends(get_db)]):
db_user = db.get(User, user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="用户不存在")
return db_user
Annotated[Session, Depends(get_db)] 得拆开看:Session 是类型,告诉编辑器 db 能调什么方法;Depends(get_db) 是个附加信息,告诉 FastAPI 调这个函数之前先把 get_db() 跑一遍,把它 yield 出来的 session 塞进 db。所以 Annotated 就是个“类型 + 附言”的打包工具,真干活的是 Depends。
get_db 之所以写成 yield 的生成器,是想让请求进来时开个 session,请求走了在 finally 里 close 掉,一个请求一个会话,不用我操心。到这儿我才第一次觉得:这框架在替我管生命周期。
分文件
路由一多,一个 app.py 就塞不下了,于是用上 APIRouter:users.py 里写 router = APIRouter(prefix="/users", tags=["users"]),app.py 里 app.include_router(users_router) 一行接进来。prefix 是给这个 router 底下所有接口统一加前缀,tags 是 /docs 里的分组名。
数据库
数据总得存下来,于是上了 SQLAlchemy。
老实说到现在我也没觉得自己算“学会”它了。我是先把这一坨照文档抄下来能跑,再回头一个一个参数问自己“这玩意到底干嘛的”。骨架大概长这样:
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
engine = create_engine("sqlite:///./app.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
先说 check_same_thread=False 这个。我一开始以为它就是个可有可无的安全开关,抄上就完事,后来才知道 SQLite 有条默认规矩:一个连接不许跨线程用。而 FastAPI 跑同步函数的时候是扔线程池里的,一扔就撞这条规矩,报错。所以只能把这个检查关掉。
但关掉不等于没事了——原来有人替你看着的线程安全,现在得你自己兜着。get_db 每个请求新建一个 session,就是在干这事儿。
SessionLocal 我卡了挺久,一直以为它就是个 session。其实不是,它是个工厂,db = SessionLocal() 一下才产出一个会话。想明白这点之后,Depends(get_db) 凭什么能一人发一份 session,才总算说得通。
剩下俩开关在 sessionmaker 里。autoflush=False 是把“每次查询之前,先把我挂起的改动刷进库”这个默认动作给关了——不关会怎样?一条还没写完的数据可能被后面的查询顺手看见,也可能提前把唯一约束撞了,反正都不是我想看到的结果。autocommit=False 更直白:自动提交关掉,你不自己 db.commit(),写的东西就一直在内存里飘着,不落盘。
这俩参数我当时纯属“教程这么写我就这么抄”,真正的意义是后来调 bug 调出来的。SQLAlchemy 给我的感觉就是这样:它本身不难,但它默认替你做了一堆决定,你得一个一个问过去,才知道自己到底让渡了什么。
CRUD
经典增删改查。就贴新建这一个,因为只有它带了我踩到的东西:
@router.post("", status_code=201, response_model=UserRead)
def create_user(payload: UserCreate, db: Annotated[Session, Depends(get_db)]):
db_user = User(
name=payload.name,
age=payload.age,
hashed_password=hash_password(payload.password),
)
db.add(db_user)
db.commit()
db.refresh(db_user) # commit 之后对象会过期,refresh 一下才能拿到自增 id
return db_user
- 路径写
""而不是"/",配prefix="/users"才是/users;写成"/"会变成/users/,跟重定向规则打架。 db.get(User, id)只适用主键查询,别的条件得db.scalars(select(User).where(...)).first()。db.refresh不能省。我一开始以为commit()完对象就直接是新的,结果拿不到自增 id,愣了半天。
当时的无语时刻:我把
get_user复制粘贴了两遍。FastAPI 一声不吭,但后定义的那个把前面的盖掉了,而且/docs里出现两条一模一样的接口——我对着文档看了半天,最后发现是自己手贱。
安全
这块我觉得是最难的一节。主要是 OAuth2 和 JWT 两套东西搅在一起:OAuth2 管“怎么要令牌”,JWT 管“令牌长啥样”。
SECRET_KEY = os.environ["SECRET_KEY"] # 从环境变量读,绝对不要硬编码提交到仓库
ALGORITHM = "HS256"
password_hash = PasswordHash.recommended()
def hash_password(plain: str) -> str:
return password_hash.hash(plain)
def verify_password(plain: str, hashed: str) -> bool:
return password_hash.verify(plain, hashed)
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
to_encode = data.copy()
to_encode["exp"] = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
代码不长,但里面埋了三个我以前混成一锅的东西,掰扯清楚才算真懂:
- HS256 是 HMAC 对称签名,保证令牌没被改过。它不是加密,payload 是明文,谁都能读。
- MD5/SHA-1 是哈希,而且早就被证明不安全了,别往密码上用。
- 密码走
pwdlib的PasswordHash.recommended()(背后是 Argon2),带随机盐的慢哈希,专门抗暴力破解。
create_access_token 干的事其实很朴素:抄一份 payload,塞个 exp(默认 15 分钟),签名编码,完事。
令牌有了总得用。登录路由:
@router.post("/login")
def login(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
db: Annotated[Session, Depends(get_db)],
):
db_user = db.scalars(select(User).where(User.name == form_data.username)).first()
if db_user is None or not verify_password(form_data.password, db_user.hashed_password):
raise HTTPException(status_code=401, detail="用户名或密码错误")
return {
"access_token": create_access_token({"sub": db_user.name}),
"token_type": "bearer",
}
坑:
login必须写在@router.get("/{user_id}")前面。路由是从上往下匹配的,一旦/{user_id}在上面,/users/login就被当成user_id="login"去转 int,直接 422。我第一次撞上还以为是密码打错了。
OAuth2PasswordRequestForm 收的是表单不是 JSON,所以得装 python-multipart。返回里那个 bearer 是 HTTP 认证方案的名字,不是令牌本身的一部分——请求头长 Authorization: Bearer <token>。
最后是验令牌,这才是最本质的东西:
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/login") # 要写真实路径,否则 /docs 的 Authorize 按钮会失败
def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: Annotated[Session, Depends(get_db)],
) -> User:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except jwt.InvalidTokenError:
raise HTTPException(
status_code=401,
detail="令牌无效或已过期",
headers={"WWW-Authenticate": "Bearer"},
) from None # from None:不把底层解码异常串进堆栈,日志里更干净
username = payload.get("sub")
if username is None:
raise HTTPException(status_code=401, detail="令牌里缺少 sub 字段")
db_user = db.scalars(select(User).where(User.name == username)).first()
if db_user is None: # 令牌合法但用户已被删除
raise HTTPException(status_code=401, detail="用户不存在")
return db_user
逻辑就三步:先试试令牌能不能解开(能解就说明没被改过、没过期),再把 sub 掏出来,最后回库里确认这人还在。所以受保护的接口只要写一行 current_user: Annotated[User, Depends(get_current_user)],鉴权就自动挂上了。
到这儿鉴权就算完事了(前端还没加,现在就是个壳子)。
几个当时以为想明白了、其实没有的地方
这几条都是写完回头才发现的,留着防以后的自己再犯:
- 更新接口我写的是
@router.put,结果里面用model_dump(exclude_unset=True)只改传上来的字段——这语义明明是 PATCH 啊。要么换patch,要么 PUT 就老老实实整体替换。当时我只想着“能跑就行”。 sub里我塞的是username。用户名哪天能改了,存量令牌就全对不上;更骚的是旧名字被别人注册,那人白捡一份鉴权。该放 user id。sqlite:///./app.db是相对当前工作目录的。换个目录起 uvicorn,它就凭空给你造个空库——我查了好一阵“数据怎么没了”。exclude_unset只挡“没传的字段”,客户端真塞个null进来照样写。想挡得再配exclude_none。
思考
导包顺序让我死磕了好久
nvim 里装了个静态检查,导包顺序不对就一直标黄。手动挪又烦得要死,而我是个见不得警告的人,碰上导包多的文件就在那儿一行一行挪,挪得怀疑人生。
最后还是靠 uv 管依赖 + 格式化工具自动排序搞定的。当时觉得白折腾,现在回头看,就是那会儿不懂“这种事该扔给工具”。
参数看不懂,硬猜了很久
自己琢磨的时候,一堆参数不知道是啥意思。特别是导包之后直接调的那种,不翻源码根本看不懂。写 JWT 那段我卡得最惨:Depends(...)、sessionmaker(...)、OAuth2PasswordBearer(...) 里面那些参数,我一个都不知道为啥这么写,最后实在没辙,靠 pi 大佬才过。
后来才慢慢养成个习惯:碰到不认识的调用,先 gd 跳到定义处瞄一眼签名和 docstring,比搜教程快多了,还不用猜。
这事对我的影响比 FastAPI 本身还大——原来“看不懂”不等于“学不会”,只是没找对看的地方。
ending
不过结局还算好。这一阶段下来,我大概算摸着 FastAPI 的门了:能自己搭一个带鉴权的 CRUD 服务,也清楚自己哪儿还不知道。
接下来想做几件事:补前端、给接口加分页和搜索、再学学部署(Docker + Nginx)这些。