从踩坑到实战:FastAPI开发Python API的完整指南

先看一个最常见的场景:你想用FastAPI写一个RESTful API,管理用户数据,比如增删改查。听起来很简单对吧?但如果你直接上手,大概率会遇到下面这些问题。

第一步:快速搭建项目骨架

FastAPI的项目结构其实很灵活,但为了后续维护方便,我建议你从开始就按模块组织代码。这是我的推荐结构:


myapi/
├── app/
│ ├── __init__.py
│ ├── main.py # 应用入口
│ ├── model

<

p>s.py # 数据模型
│ ├── schemas.py # Pydantic模型(请求/响应)
│ ├── crud.py # 数据库操作
│ ├── database.py # 数据库连接
│ └── routers/
│ ├── __init__.py
│ └── users.py # 用户路由
├── requirements.txt
└── .env
`

千万别把所有代码塞进一个文件里,除非你只想写个demo。我一开始就犯了这个错,结果一个月后想加新功能,看到那3000行的代码直接崩溃。

核心:定义数据模型和API端点

先看数据库模型和Pydantic模型。为什么要分开?因为数据库模型是ORM的,而Pydantic模型是用来做数据验证和序列化的,两者职责不同。

`python

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

用SQLite做demo,生产环境换PostgreSQL

SQLALCHEMY_DATABASE_URL = "sqlite:///./myapi.db"

engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
`

`python

models.py

from sqlalchemy import Column, Integer, String,

Boolean, DateTime
from sqlalchemy.sql import func
from .database import Base

class User(Base):
__tablename__ = "users"

id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
username = Column(String, unique=True, index=True, nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
`

这里有个坑:SQLAlchemy的server_defaultdefault不一样。server_default是在数据库层面生成默认值,而default是Python层面。如果你用异步框架,最好用server_default,避免Python层面的时间不一致。

关键:Pydantic模型与CRUD

再看Pydantic模型,这是FastAPI的灵魂。为什么不用SQLAlchemy直接返回?因为Pydantic可以做数据验证、嵌套校验、自动生成文档。

`python

schemas.py

from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional

class UserBase(BaseModel):
email: EmailStr
username: str

class UserCreate(UserBase):
password: str # 生产环境要hash

class UserUpdate(BaseModel):
email: Optional[EmailStr] = None
username: Optional[str] = None
is_active: Optional[bool] = None

class UserResponse(UserBase):
id: int
is_active: bool
created_at: datetime

class Config:
from_attributes = True # 允许从ORM对象创建
`

注意这个from_attributes = True,在Pydantic v2中替代了旧版的orm_mode = True。官方文档写得很绕,我看了三遍才明白,其实就是允许从SQLAlchemy对象直接创建Pydantic模型。

接下来是CRUD操作:

`python

crud.py

from sqlalchemy.orm import Session
from . import models, schemas

def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
# 这里简化处理,实际要hash密码
db_user = models.User(
email=user.email,
username=user.username
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user

def update_user(db: Session, user_id: int, user: schemas.UserUpdate):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if db_user is None:
return None
update_data = user.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(db_user, key, value)
db.commit()
db.refresh(db_user)
return db_user
`

这里有个设计上的反人类之处:exclude_unset=True。如果你不传这个参数,Pydantic会把所有字段都当成None发送到数据库,导致你只想更新用户名时,邮箱也被清空了。我第一次用的时候debug了两小时才发现。

路由:构建API端点

`python

routers/users.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from .. import schemas, crud
from ..database import SessionLocal

router = APIRouter(
prefix="/users",
tags=["users"],
responses={404: {"description": "Not found"}}
)

依赖注入:获取数据库会话

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

@router.post("/", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
return crud.create_user(db=db, user=user)

@router.get("/", response_model=List[schemas.UserResponse])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users

@router.get("/{user_id}", response_model=schemas.UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user

@router.put("/{user_id}", response_model=schemas.UserResponse)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
db_user = crud.update_user(db=db, user_id=user_id, user=user)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
`

另一个坑:依赖注入的使用场景

FastAPI的依赖注入系统确实强大,但很多人用错了。它的真正价值不是"传数据库会话",而是"解耦业务逻辑"。

比如权限校验,你可以在每个路由里写if语句,但更好的做法是:

`python

dependencies.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
return username

在路由中使用

@router.get("/me", response_model=schemas.UserResponse)
def read_current_user(current_user: str = Depends(get_current_user)):
# 直接使用current_user,不用在函数里写校验逻辑
return current_user
`

这个设计真的反人类?其实是反直觉。但一旦习惯,你会发现代码清爽很多。官方文档写这段的时候跟谜语一样,我建议你看实战代码而不是官方说明。

性能优化:异步与并发

FastAPI最大的卖点是异步。但很多人以为只要用async就自动快了,这是误区。

`python

错误的做法

@app.get("/slow")
async def slow_endpoint():
# 同步代码被async包裹,毫无意义
time.sleep(2) # 同步阻塞
return {"message": "slow"}

正确的做法

@app.get("/fast")
async def fast_endpoint():
# 异步I/O操作
await asyncio.sleep(2) # 异步非阻塞
return {"message": "fast"}
`

如果你用同步的time.sleep,即使函数声明为async,它依然是阻塞的。真正提升性能的是异步I/O操作,比如数据库查询用异步驱动(asyncpg + SQLAlchemy 1.4+ async)。

我做过一个测试:同样的API,同步写法在100并发下耗时从3.2秒降到0.8秒,但前提是你真的用了异步数据库驱动。

还有个技巧:自动生成API文档

FastAPI自动帮你生成Swagger文档和ReDoc,但默认可能不够友好。你可以自定义:

`python
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI()

def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="My API - 用户管理系统",
version="1.0.0",
description="这是一个实战API,支持用户CRUD操作",
routes=app.routes,
)
openapi_schema["info"]["x-logo"] = {
"url": "https://example.com/logo.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema

app.openapi = custom_openapi
`

这样你的API文档看起来更专业,给前端同事看的时候也更有面子。

部署与生产环境注意事项

最后说部署。别用自带的uvicorn跑生产,至少用个进程管理器。我的推荐配置:

`bash

使用gunicorn + uvicorn workers

gunicorn app.main:app \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--workers 4 \
--timeout 120 \
--access-logfile -
`

为什么用4个worker?因为通常CPU核心数*2+1是最优的。但如果你用异步数据库,一个worker就可以处理很多请求,不用盲目加worker。

总结一下,你可以立刻用的三个点

  • 项目结构:从一开始就按模块拆分,不要把所有代码塞一个文件。用models.pyschemas.pycrud.py`三层分离,后续加功能就像搭积木。
  • 依赖注入:不要只在路由里传数据库会话,把权限校验、日志记录、缓存检查都做成依赖,让业务逻辑更干净。
  • 异步是真异步:如果要用异步,确保你的数据库驱动、HTTP客户端都是异步的。不然只是给函数加个async,性能没提升反而有开销。
  • 踩坑无数后,我真心觉得FastAPI是Python Web框架里最舒服的。只要跳过那些坑,它能让你的API开发效率提升50%以上。如果你也踩过什么坑,欢迎在评论区分享,大家一起进步。

    滚动至顶部