当前位置：首页 > 学习笔记 > 正文内容

FastAPI 后端开发实战完全指南：从入门到精通

廖万里16小时前学习笔记1

FastAPI 是当前最热门的 Python Web 框架，以其高性能、易用性和自动生成 API 文档的特性，成为后端开发者的首选。本文将带你从零开始，系统掌握 FastAPI 的核心概念、路由系统、依赖注入、数据库操作、认证授权等实战技能。

一、核心概念

FastAPI 是一个现代、高性能的 Python Web 框架，基于 Starlette 和 Pydantic 构建。它具有以下核心优势：

1. 高性能：与 Node.js 和 Go 性能相当，是最快的 Python 框架之一

2. 自动文档：基于 OpenAPI 和 JSON Schema 自动生成交互式 API 文档

3. 类型提示：充分利用 Python 类型注解，提供编辑器支持和自动校验

4. 异步支持：原生支持 async/await，轻松处理高并发场景

5. 依赖注入：强大且简单的依赖注入系统

# 安装 FastAPI 和 Uvicorn（ASGI 服务器）# pip install fastapi uvicorn# 最简单的 FastAPI 应用from fastapi import FastAPI# 创建 FastAPI 实例app = FastAPI(    title="我的第一个 API",    description="学习 FastAPI 的示例项目",    version="1.0.0")# 定义一个 GET 路由@app.get("/")async def root():    """根路径，返回欢迎信息"""    return {"message": "Hello, FastAPI!"}# 运行服务器: uvicorn main:app --reload

二、路由系统详解

FastAPI 的路由系统支持所有 HTTP 方法，并提供了强大的参数校验能力。

2.1 路径参数

from fastapi import FastAPIfrom typing import Optionalapp = FastAPI()# 基本路径参数@app.get("/users/{user_id}")async def get_user(user_id: int):    """    获取用户信息    user_id: int - 自动校验必须是整数    """    return {"user_id": user_id}# 枚举类型的路径参数from enum import Enumclass ModelName(str, Enum):    alexnet = "alexnet"    resnet = "resnet"    lenet = "lenet"@app.get("/models/{model_name}")async def get_model(model_name: ModelName):    """使用枚举限制可选值"""    if model_name == ModelName.alexnet:        return {"model": model_name, "message": "Deep Learning FTW!"}    return {"model": model_name}# 路径参数包含路径@app.get("/files/{file_path:path}")async def read_file(file_path: str):    """处理包含斜杠的路径参数"""    return {"file_path": file_path}

2.2 查询参数

from fastapi import FastAPI, Queryfrom typing import Optional, Listapp = FastAPI()# 基本查询参数@app.get("/items/")async def read_items(    skip: int = 0,           # 默认值    limit: int = 10,         # 默认值    q: Optional[str] = None  # 可选参数):    """    查询物品列表    skip: 跳过的记录数    limit: 返回的最大记录数    q: 搜索关键词（可选）    """    return {"skip": skip, "limit": limit, "q": q}# 参数校验@app.get("/items/{item_id}")async def read_item(    item_id: int,    q: Optional[str] = Query(        None,                               # 默认值        min_length=3,                       # 最小长度        max_length=50,                      # 最大长度        regex="^[a-zA-Z0-9_-]+$",           # 正则校验        description="搜索关键词"             # 文档描述    )):    """带校验的查询参数"""    return {"item_id": item_id, "q": q}# 多值查询参数@app.get("/users/")async def read_users(    ids: List[int] = Query([], description="用户ID列表")):    """支持多个值的查询参数: /users/?ids=1&ids=2&ids=3"""    return {"ids": ids}

2.3 请求体

from fastapi import FastAPIfrom pydantic import BaseModel, Field, validatorfrom typing import Optional, Listfrom datetime import datetimeapp = FastAPI()# 定义请求体模型class Item(BaseModel):    name: str = Field(..., min_length=1, max_length=100, description="物品名称")    description: Optional[str] = Field(None, max_length=500)    price: float = Field(..., gt=0, description="价格必须大于0")    tax: Optional[float] = None    tags: List[str] = []        # 自定义验证器    @validator("name")    def name_must_not_contain_special_chars(cls, v):        if any(char in v for char in ["@", "#", "$"]):            raise ValueError("名称不能包含特殊字符")        return v        class Config:        # 示例数据，用于 API 文档        schema_extra = {            "example": {                "name": "iPhone 15",                "description": "最新款智能手机",                "price": 6999.00,                "tax": 0.1,                "tags": ["电子产品", "手机"]            }        }# 使用请求体@app.post("/items/")async def create_item(item: Item):    """    创建物品    自动校验请求体是否符合 Item 模型    """    return {        "message": "创建成功",        "item": item,        "price_with_tax": item.price + (item.tax or 0) * item.price    }

三、依赖注入系统

FastAPI 的依赖注入系统是其最强大的特性之一，可以轻松实现代码复用和解耦。

from fastapi import FastAPI, Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearerfrom typing import Optionalapp = FastAPI()# 简单依赖def common_parameters(    q: Optional[str] = None,    skip: int = 0,    limit: int = 100):    """通用查询参数"""    return {"q": q, "skip": skip, "limit": limit}@app.get("/items/")async def read_items(commons: dict = Depends(common_parameters)):    """使用依赖注入获取通用参数"""    return commons# 类作为依赖class CommonQueryParams:    def __init__(        self,        q: Optional[str] = None,        skip: int = 0,        limit: int = 100    ):        self.q = q        self.skip = skip        self.limit = limit@app.get("/users/")async def read_users(commons: CommonQueryParams = Depends()):    """使用类作为依赖，自动实例化"""    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}# OAuth2 依赖oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")@app.get("/users/me")async def read_users_me(token: str = Depends(oauth2_scheme)):    """获取当前用户信息"""    return {"token": token}

四、数据库操作

FastAPI 可以与 SQLAlchemy、Tortoise ORM 等数据库工具完美配合。这里以 SQLAlchemy 为例：

# 安装依赖: pip install sqlalchemy asyncpgfrom fastapi import FastAPI, Depends, HTTPExceptionfrom sqlalchemy import create_engine, Column, Integer, String, Booleanfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import Session, sessionmakerfrom pydantic import BaseModelfrom typing import Optional# 数据库配置SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"engine = create_engine(SQLALCHEMY_DATABASE_URL)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()# 定义数据库模型class UserDB(Base):    __tablename__ = "users"        id = Column(Integer, primary_key=True, index=True)    username = Column(String, unique=True, index=True)    email = Column(String, unique=True, index=True)    hashed_password = Column(String)    is_active = Column(Boolean, default=True)# 创建表Base.metadata.create_all(bind=engine)# Pydantic 模型（用于 API）class UserCreate(BaseModel):    username: str    email: str    password: strclass UserResponse(BaseModel):    id: int    username: str    email: str    is_active: bool        class Config:        orm_mode = True# 数据库依赖def get_db():    """获取数据库会话"""    db = SessionLocal()    try:        yield db    finally:        db.close()# FastAPI 应用app = FastAPI()@app.post("/users/", response_model=UserResponse)def create_user(user: UserCreate, db: Session = Depends(get_db)):    """创建用户"""    db_user = db.query(UserDB).filter(UserDB.email == user.email).first()    if db_user:        raise HTTPException(status_code=400, detail="邮箱已被注册")        fake_hashed_password = user.password + "notreallyhashed"    db_user = UserDB(        username=user.username,        email=user.email,        hashed_password=fake_hashed_password    )    db.add(db_user)    db.commit()    db.refresh(db_user)    return db_user

五、认证与授权

FastAPI 内置了对 OAuth2 和 JWT 的支持，可以轻松实现安全的认证系统。

from fastapi import FastAPI, Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormfrom jose import JWTError, jwtfrom passlib.context import CryptContextfrom pydantic import BaseModelfrom datetime import datetime, timedeltafrom typing import Optional# 配置SECRET_KEY = "your-secret-key-keep-it-secret"ALGORITHM = "HS256"ACCESS_TOKEN_EXPIRE_MINUTES = 30# 密码加密pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")# OAuth2 方案oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")app = FastAPI()# 模拟用户数据库fake_users_db = {    "johndoe": {        "username": "johndoe",        "email": "johndoe@example.com",        "hashed_password": pwd_context.hash("secret"),        "disabled": False,    }}# Pydantic 模型class Token(BaseModel):    access_token: str    token_type: strclass User(BaseModel):    username: str    email: Optional[str] = None    disabled: Optional[bool] = None# JWT 工具函数def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):    """创建 JWT token"""    to_encode = data.copy()    if expires_delta:        expire = datetime.utcnow() + expires_delta    else:        expire = datetime.utcnow() + timedelta(minutes=15)    to_encode.update({"exp": expire})    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)    return encoded_jwt# API 路由@app.post("/token", response_model=Token)async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):    """用户登录，获取 token"""    user = authenticate_user(fake_users_db, form_data.username, form_data.password)    if not user:        raise HTTPException(            status_code=status.HTTP_401_UNAUTHORIZED,            detail="用户名或密码错误",            headers={"WWW-Authenticate": "Bearer"},        )    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    access_token = create_access_token(        data={"sub": user.username}, expires_delta=access_token_expires    )    return {"access_token": access_token, "token_type": "bearer"}@app.get("/users/me")async def read_users_me(current_user: User = Depends(get_current_active_user)):    """获取当前用户信息（需要认证）"""    return current_user

六、中间件与异常处理

from fastapi import FastAPI, Request, statusfrom fastapi.responses import JSONResponsefrom fastapi.middleware.cors import CORSMiddlewarefrom fastapi.middleware.gzip import GZipMiddlewareimport timeapp = FastAPI()# CORS 中间件app.add_middleware(    CORSMiddleware,    allow_origins=["*"],    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],)# Gzip 压缩中间件app.add_middleware(GZipMiddleware, minimum_size=1000)# 自定义中间件@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)    return response# 自定义异常处理@app.exception_handler(RequestValidationError)async def validation_exception_handler(request: Request, exc: RequestValidationError):    """自定义验证错误响应"""    return JSONResponse(        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,        content={            "message": "请求参数验证失败",            "errors": exc.errors()        },    )

七、异步编程最佳实践

import asynciofrom fastapi import FastAPI, BackgroundTasksfrom typing import Listimport httpxapp = FastAPI()# 异步 HTTP 客户端async def fetch_url(url: str):    """异步获取 URL 内容"""    async with httpx.AsyncClient() as client:        response = await client.get(url)        return response.text# 并发请求@app.get("/fetch-multiple")async def fetch_multiple():    """并发获取多个 URL"""    urls = [        "https://httpbin.org/delay/1",        "https://httpbin.org/delay/2",        "https://httpbin.org/delay/3",    ]    tasks = [fetch_url(url) for url in urls]    results = await asyncio.gather(*tasks)    return {"results_count": len(results)}# 后台任务def send_email(email: str, message: str):    """发送邮件（后台任务）"""    import time    time.sleep(2)    print(f"邮件已发送到 {email}: {message}")@app.post("/send-notification/{email}")async def send_notification(    email: str,    background_tasks: BackgroundTasks):    """发送通知（后台任务）"""    background_tasks.add_task(        send_email,        email,        "您有一条新消息！"    )    return {"message": "通知已加入发送队列", "email": email}

八、部署与优化

# 使用 Gunicorn + Uvicorn 部署gunicorn main:app \    --workers 4 \    --worker-class uvicorn.workers.UvicornWorker \    --bind 0.0.0.0:8000# Docker 部署# DockerfileFROM python:3.11-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY . .CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000"]