FastAPI 完全指南 2026：从零构建高性能异步 Python API 到 Docker 生产部署

1
from fastapi import Depends, HTTPException, status
2
from fastapi.security import OAuth2PasswordBearer
3
from jose import JWTError, jwt
4
from .config import settings
5
from .database import AsyncSessionLocal
6

7
# ── 数据库 Session 依赖 ───────────────────────────────────────
8
async def get_db():
9
    async with AsyncSessionLocal() as session:
10
        try:
11
            yield session
12
            await session.commit()
13
        except Exception:
14
            await session.rollback()
15
            raise
16

17
# ── JWT 认证依赖 ──────────────────────────────────────────────
18
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
19

20
async def get_current_user(token: str = Depends(oauth2_scheme)):
21
    credentials_exception = HTTPException(
22
        status_code=status.HTTP_401_UNAUTHORIZED,
23
        detail="无效的认证凭证",
24
        headers={"WWW-Authenticate": "Bearer"},
25
    )
26
    try:
27
        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
28
        user_id: int | None = payload.get("sub")
29
        if user_id is None:
30
            raise credentials_exception
31
    except JWTError:
32
        raise credentials_exception
33
    return user_id
34

35
# ── 在路由中使用依赖 ──────────────────────────────────────────
36
from fastapi import APIRouter
37
from sqlalchemy.ext.asyncio import AsyncSession
38

39
router = APIRouter()
40

41
@router.get("/me")
42
async def get_me(
43
    current_user_id: int = Depends(get_current_user),
44
    db: AsyncSession = Depends(get_db),
45
):
46
    # current_user_id 和 db 自动注入
47
    return {"user_id": current_user_id}

1
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
2
from sqlalchemy.orm import DeclarativeBase
3
from .config import settings
4

5
engine = create_async_engine(
6
    settings.database_url,
7
    echo=settings.debug,     # debug=True 时打印 SQL 语句
8
    pool_size=10,
9
    max_overflow=20,
10
)
11

12
AsyncSessionLocal = async_sessionmaker(
13
    engine,
14
    class_=AsyncSession,
15
    expire_on_commit=False,
16
)
17

18
class Base(DeclarativeBase):
19
    pass

1
from datetime import datetime
2
from sqlalchemy import String, Boolean, DateTime, func
3
from sqlalchemy.orm import Mapped, mapped_column
4
from ..database import Base
5

6
class User(Base):
7
    __tablename__ = "users"
8

9
    id:         Mapped[int]      = mapped_column(primary_key=True, index=True)
10
    username:   Mapped[str]      = mapped_column(String(50), unique=True, index=True)
11
    email:      Mapped[str]      = mapped_column(String(255), unique=True, index=True)
12
    hashed_password: Mapped[str] = mapped_column(String(255))
13
    is_active:  Mapped[bool]     = mapped_column(Boolean, default=True)
14
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
15
    updated_at: Mapped[datetime] = mapped_column(
16
        DateTime, server_default=func.now(), onupdate=func.now()
17
    )

1
from sqlalchemy import select
2
from sqlalchemy.ext.asyncio import AsyncSession
3
from ..models.user import User
4
from ..schemas.user import UserCreate
5
from passlib.context import CryptContext
6

7
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
8

9
class UserService:
10
    def __init__(self, db: AsyncSession):
11
        self.db = db
12

13
    async def get_by_id(self, user_id: int) -> User | None:
14
        result = await self.db.execute(select(User).where(User.id == user_id))
15
        return result.scalar_one_or_none()
16

17
    async def get_by_email(self, email: str) -> User | None:
18
        result = await self.db.execute(select(User).where(User.email == email))
19
        return result.scalar_one_or_none()
20

21
    async def create(self, data: UserCreate) -> User:
22
        user = User(
23
            username=data.username,
24
            email=data.email,
25
            hashed_password=pwd_context.hash(data.password),
26
        )
27
        self.db.add(user)
28
        await self.db.flush()   # 获取自动生成的 ID，不提交事务
29
        await self.db.refresh(user)
30
        return user
31

32
    def verify_password(self, plain: str, hashed: str) -> bool:
33
        return pwd_context.verify(plain, hashed)

1
from datetime import datetime, timedelta, timezone
2
from fastapi import APIRouter, Depends, HTTPException, status
3
from fastapi.security import OAuth2PasswordRequestForm
4
from jose import jwt
5
from sqlalchemy.ext.asyncio import AsyncSession
6
from ..config import settings
7
from ..deps import get_db
8
from ..services.user import UserService
9

10
router = APIRouter(prefix="/auth", tags=["认证"])
11

12
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
13
    to_encode = data.copy()
14
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=30))
15
    to_encode.update({"exp": expire})
16
    return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
17

18
@router.post("/login")
19
async def login(
20
    form_data: OAuth2PasswordRequestForm = Depends(),
21
    db: AsyncSession = Depends(get_db),
22
):
23
    service = UserService(db)
24
    user = await service.get_by_email(form_data.username)
25

26
    if not user or not service.verify_password(form_data.password, user.hashed_password):
27
        raise HTTPException(
28
            status_code=status.HTTP_401_UNAUTHORIZED,
29
            detail="邮箱或密码错误",
30
        )
31

32
    access_token = create_access_token(
33
        data={"sub": str(user.id)},
34
        expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
35
    )
36
    return {
37
        "access_token": access_token,
38
        "token_type": "bearer",
39
    }

1
import time
2
import uuid
3
from fastapi import FastAPI, Request
4
from fastapi.middleware.cors import CORSMiddleware
5
from fastapi.responses import JSONResponse
6
from .config import settings
7
from .routers import users, auth
8

9
app = FastAPI(title=settings.app_name)
10

11
# ── CORS 跨域中间件 ───────────────────────────────────────────
12
app.add_middleware(
13
    CORSMiddleware,
14
    allow_origins=["https://your-frontend.com"],  # 生产环境写具体域名
15
    allow_credentials=True,
16
    allow_methods=["*"],
17
    allow_headers=["*"],
18
)
19

20
# ── 请求 ID 和耗时日志中间件 ──────────────────────────────────
21
@app.middleware("http")
22
async def add_request_context(request: Request, call_next):
23
    request_id = str(uuid.uuid4())[:8]
24
    start = time.perf_counter()
25

26
    response = await call_next(request)
27

28
    duration = (time.perf_counter() - start) * 1000
29
    response.headers["X-Request-ID"] = request_id
30
    response.headers["X-Process-Time"] = f"{duration:.2f}ms"
31
    return response
32

33
# ── 全局异常处理 ──────────────────────────────────────────────
34
@app.exception_handler(404)
35
async def not_found_handler(request: Request, exc):
36
    return JSONResponse(status_code=404, content={"error": "资源不存在"})
37

38
@app.exception_handler(500)
39
async def server_error_handler(request: Request, exc):
40
    return JSONResponse(status_code=500, content={"error": "服务器内部错误"})
41

42
# ── 注册路由 ──────────────────────────────────────────────────
43
app.include_router(auth.router, prefix=settings.api_prefix)
44
app.include_router(users.router, prefix=settings.api_prefix)

1
import pytest
2
import pytest_asyncio
3
from httpx import AsyncClient, ASGITransport
4
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
5
from src.my_api.main import app
6
from src.my_api.database import Base
7
from src.my_api.deps import get_db
8

9
# 使用内存 SQLite 作为测试数据库
10
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
11

12
@pytest_asyncio.fixture(scope="session")
13
async def db_engine():
14
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
15
    async with engine.begin() as conn:
16
        await conn.run_sync(Base.metadata.create_all)
17
    yield engine
18
    await engine.dispose()
19

20
@pytest_asyncio.fixture
21
async def db_session(db_engine):
22
    session_factory = async_sessionmaker(db_engine, class_=AsyncSession)
23
    async with session_factory() as session:
24
        yield session
25
        await session.rollback()  # 每个测试后回滚，保持隔离
26

27
@pytest_asyncio.fixture
28
async def client(db_session):
29
    # 覆盖依赖注入，使用测试数据库
30
    app.dependency_overrides[get_db] = lambda: db_session
31

32
    async with AsyncClient(
33
        transport=ASGITransport(app=app),
34
        base_url="http://test",
35
    ) as c:
36
        yield c
37

38
    app.dependency_overrides.clear()

1
import pytest
2

3
@pytest.mark.asyncio
4
async def test_create_user(client):
5
    response = await client.post("/api/v1/users", json={
6
        "username": "testuser",
7
        "email": "test@example.com",
8
        "password": "SecurePass123!",
9
        "age": 25,
10
    })
11
    assert response.status_code == 201
12
    data = response.json()
13
    assert data["username"] == "testuser"
14
    assert "password" not in data   # 确保密码不在响应中
15

16
@pytest.mark.asyncio
17
async def test_login(client):
18
    # 先创建用户
19
    await client.post("/api/v1/users", json={
20
        "username": "loginuser",
21
        "email": "login@example.com",
22
        "password": "TestPass123!",
23
        "age": 30,
24
    })
25

26
    # 测试登录
27
    response = await client.post(
28
        "/api/v1/auth/login",
29
        data={"username": "login@example.com", "password": "TestPass123!"},
30
    )
31
    assert response.status_code == 200
32
    assert "access_token" in response.json()
33

34
@pytest.mark.asyncio
35
async def test_get_me_unauthorized(client):
36
    response = await client.get("/api/v1/me")
37
    assert response.status_code == 401