第 11 章: 认证

本章概述

学习如何在FastAPI中实现用户认证，包括注册、登录、令牌生成和注销。

11.1 基础认证

使用HTTP Basic Auth：

from fastapi import Depends, HTTPException
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

@app.get("/users/me")
def read_current_user(credentials: HTTPBasicCredentials = Depends(security)):
    return {"username": credentials.username}

11.2 OAuth2密码Bearer认证

FastAPI提供了OAuth2密码Bearer认证的支持：

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

11.3 数据库认证

完整的用户认证系统：

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

router = APIRouter()

@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    # 验证用户名和密码
    user = db.query(User).filter(User.username == form_data.username).first()
    if not user or not verify_password(form_data.password, user.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password"
        )
    
    # 生成访问令牌
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

11.4 保护路由

使用依赖项保护API方法：

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    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
    
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception
    return user

@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

11.5 注销

@app.post("/logout")
def logout(response: Response):
    response.delete_cookie("session")
    return {"message": "Logged out successfully"}

认证流程图1 认证流程图2 令牌验证保护路由示例

小结

在本章中，我们学习了： - ✅ 实现基础认证 - ✅ 使用OAuth2和密码Bearer - ✅ 生成和验证访问令牌 - ✅ 保护API路由 - ✅ 实现注销功能

在下一章中，我们将学习注解和返回类型！

第 12 章: 注解、省略号(…)表示法和返回类型

本章概述

学习FastAPI中的类型注解、Ellipsis表示法和返回类型声明。

12.1 Ellipsis表示法

... (Ellipsis) 表示必填参数：

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
def read_items(q: str = Query(...)):  # q是必填的
    return {"q": q}

12.2 Annotated类型

Python 3.9+ 支持Annotated：

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
def read_items(q: Annotated[str, Query(min_length=3)]):
    return {"q": q}

12.3 返回类型

声明返回类型用于生成文档：

from typing import List
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float

@app.get("/items/", response_model=List[Item])
def read_items():
    return [
        {"name": "Item 1", "price": 10.5},
        {"name": "Item 2", "price": 20.0}
    ]

12.4 response_model的作用

过滤返回数据（只返回模型中定义的字段）
生成API文档
验证返回数据

小结

在本章中，我们学习了： - ✅ 使用Ellipsis (…) 表示必填参数 - ✅ 使用Annotated类型 - ✅ 声明返回类型用于文档生成 - ✅ 理解response_model的作用

第 13 章: FastAPI应用测试

本章概述

学习如何使用pytest为FastAPI应用编写单元测试。

13.1 pytest简介

pytest是Python的测试框架：

pip install pytest
pip install httpx  # 用于测试API

13.2 创建第一个测试

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

13.3 使用pytest fixtures删除重复代码

import pytest

@pytest.fixture
def client():
    from main import app
    with TestClient(app) as c:
        yield c

def test_read_main(client):
    response = client.get("/")
    assert response.status_code == 200

13.4 测试用户模块

def test_create_user(client):
    response = client.post(
        "/users/",
        json={"username": "test", "password": "test123"}
    )
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "test"

def test_login(client):
    response = client.post(
        "/token",
        data={"username": "test", "password": "test123"}
    )
    assert response.status_code == 200
    assert "access_token" in response.json()

13.5 测试任务模块

def test_create_task(client):
    # 先登录获取token
    response = client.post(
        "/token",
        data={"username": "test", "password": "test123"}
    )
    token = response.json()["access_token"]
    
    # 创建任务
    response = client.post(
        "/tasks/",
        json={"name": "Test Task"},
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 201

13.6 测试数据库

使用测试数据库：

@pytest.fixture
def db():
    # 创建测试数据库
    Base.metadata.create_all(bind=test_engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=test_engine)

小结

在本章中，我们学习了： - ✅ 使用pytest编写测试 - ✅ 使用TestClient测试API - ✅ 使用fixtures管理测试依赖 - ✅ 测试用户认证 - ✅ 测试CRUD操作 - ✅ 使用测试数据库

总结

恭喜！你已经完成了FastAPI的学习之路。现在你应该能够： - ✅ 创建FastAPI应用 - ✅ 定义路由和参数 - ✅ 使用Pydantic验证数据 - ✅ 连接数据库 - ✅ 使用模板引擎 - ✅ 实现认证 - ✅ 编写测试

继续实践和探索，祝你在FastAPI的开发旅程中取得成功！