# FastAPI 在电商系统开发中的最佳实践

## 引言

在当今快速发展的电商行业中，后端 API 的性能、可维护性和开发效率直接影响着用户体验和业务增长。FastAPI 作为一个现代、快速（高性能）的 Python Web 框架，凭借其出色的性能、直观的类型提示和自动 API 文档生成等特性，已成为构建电商后端服务的理想选择。

本文将深入探讨如何在实际电商项目中应用 FastAPI 的最佳实践，涵盖项目结构、依赖注入、异步处理、数据验证、安全认证等关键方面。无论您是正在构建一个新的电商平台，还是优化现有系统，这些实践都将帮助您创建更健壮、可扩展和高效的 API 服务。

## 核心内容

### 1. 项目结构设计

良好的项目结构是维护大型电商应用的基础。以下是一个推荐的电商项目结构：

```
ecommerce-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用入口
│   ├── core/               # 核心配置和工具
│   │   ├── __init__.py
│   │   ├── config.py       # 配置管理
│   │   ├── security.py     # 安全相关工具
│   │   └── dependencies.py # 依赖注入
│   ├── api/               # API 路由
│   │   ├── __init__.py
│   │   ├── v1/            # API 版本管理
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│   │   │   │   ├── auth.py
│   │   │   │   ├── products.py
│   │   │   │   ├── orders.py
│   │   │   │   └── users.py
│   │   │   └── api.py     # API 路由注册
│   ├── models/            # 数据模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── product.py
│   │   └── order.py
│   ├── schemas/           # Pydantic 模式
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── product.py
│   │   └── order.py
│   ├── services/          # 业务逻辑层
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   ├── product_service.py
│   │   └── order_service.py
│   └── db/               # 数据库相关
│       ├── __init__.py
│       ├── session.py    # 数据库会话管理
│       └── repositories/ # 数据访问层
│           ├── __init__.py
│           ├── user_repo.py
│           └── product_repo.py
├── tests/                # 测试文件
├── alembic/             # 数据库迁移
├── requirements.txt
└── .env.example
```

### 2. 配置管理

电商系统通常需要多种环境配置（开发、测试、生产）。使用 Pydantic 的 `BaseSettings` 可以优雅地管理配置：

```python
# app/core/config.py
from pydantic import BaseSettings, PostgresDsn, validator
from typing import Optional, List

class Settings(BaseSettings):
    # 基础配置
    API_V1_STR: str = "/api/v1"
    PROJECT_NAME: str = "E-Commerce API"
    
    # 安全配置
    SECRET_KEY: str
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    # 数据库配置
    POSTGRES_SERVER: str
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_DB: str
    DATABASE_URI: Optional[PostgresDsn] = None
    
    # CORS 配置
    BACKEND_CORS_ORIGINS: List[str] = []
    
    @validator("DATABASE_URI", pre=True)
    def assemble_db_connection(cls, v: Optional[str], values: dict) -> str:
        if isinstance(v, str):
            return v
        
        return PostgresDsn.build(
            scheme="postgresql",
            user=values.get("POSTGRES_USER"),
            password=values.get("POSTGRES_PASSWORD"),
            host=values.get("POSTGRES_SERVER"),
            path=f"/{values.get('POSTGRES_DB') or ''}",
        )
    
    @validator("BACKEND_CORS_ORIGINS", pre=True)
    def assemble_cors_origins(cls, v: str | List[str]) -> List[str]:
        if isinstance(v, str) and not v.startswith("["):
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, (list, str)):
            return v
        raise ValueError(v)
    
    class Config:
        env_file = ".env"
        case_sensitive = True

settings = Settings()
```

### 3. 依赖注入系统

FastAPI 的依赖注入系统是其强大功能之一，特别适合电商系统中的复杂业务逻辑：

```python
# app/core/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session

from app.core.config import settings
from app.db.session import get_db
from app.models.user import User
from app.schemas.token import TokenPayload

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl=f"{settings.API_V1_STR}/auth/login"
)

async def get_current_user(
    db: Session = Depends(get_db),
    token: str = Depends(oauth2_scheme)
) -> User:
    """获取当前认证用户"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭证",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = jwt.decode(
            token, 
            settings.SECRET_KEY, 
            algorithms=[settings.ALGORITHM]
        )
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
        token_data = TokenPayload(sub=user_id)
    except JWTError:
        raise credentials_exception
    
    user = db.query(User).filter(User.id == token_data.sub).first()
    if user is None:
        raise credentials_exception
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户已被禁用"
        )
    
    return user

def get_current_active_superuser(
    current_user: User = Depends(get_current_user),
) -> User:
    """检查当前用户是否为超级管理员"""
    if not current_user.is_superuser:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="权限不足"
        )
    return current_user
```

### 4. 异步数据库操作

电商系统需要处理高并发请求，异步操作至关重要：

```python
# app/db/session.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

# 创建异步引擎
engine = create_async_engine(
    settings.DATABASE_URI.replace("postgresql", "postgresql+asyncpg"),
    echo=True,  # 生产环境设为 False
    future=True
)

# 创建异步会话工厂
AsyncSessionLocal = sessionmaker(
    engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)

async def get_db() -> AsyncSession:
    """获取数据库会话依赖"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()
```

### 5. 数据验证与序列化

使用 Pydantic 模型进行数据验证和序列化：

```python
# app/schemas/product.py
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from decimal import Decimal

class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    price: Decimal = Field(..., gt=0)
    stock_quantity: int = Field(..., ge=0)
    category_id: int
    sku: str = Field(..., min_length=1, max_length=50)
    
    @validator("price")
    def validate_price(cls, v):
        """验证价格格式"""
        if v.as_tuple().exponent < -2:
            raise ValueError("价格最多支持两位小数")
        return v

class ProductCreate(ProductBase):
    pass

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    price: Optional[Decimal] = Field(None, gt=0)
    stock_quantity: Optional[int] = Field(None, ge=0)
    category_id: Optional[int] = None

class ProductInDB(ProductBase):
    id: int
    created_at: datetime
    updated_at: datetime
    is_active: bool
    
    class Config:
        orm_mode = True

class ProductResponse(ProductInDB):
    pass

class ProductListResponse(BaseModel):
    items: List[ProductResponse]
    total: int
    page: int
    size: int
    pages: int
```

## 实践应用

### 1. 商品管理 API 实现

```python
# app/api/v1/endpoints/products.py
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func

from app.core.dependencies import get_current_user, get_current_active_superuser
from app.db.session import get_db
from app.models.product import Product
from app.models.user import User
from app.schemas.product import (
    ProductCreate, 
    ProductUpdate, 
    ProductResponse,
    ProductListResponse
)
from app.services.product_service import ProductService

router = APIRouter()

@router.get("/", response_model=ProductListResponse)
async def list_products(
    db: AsyncSession = Depends(get_db),
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    category_id: Optional[int] = None,
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    search: Optional[str] = None,
    in_stock_only: bool = False
):
    """
    获取商品列表
    - 支持分页
    - 支持按分类筛选
    - 支持价格区间筛选
    - 支持关键词搜索
    - 支持仅显示有库存商品
    """
    # 构建查询条件
    conditions = []
    
    if category_id:
        conditions.append(Product.category_id == category_id)
    
    if min_price is not None:
        conditions.append(Product.price >= min_price)
    
    if max_price is not None:
        conditions.append(Product.price <= max_price)
    
    if in_stock_only:
        conditions.append(Product.stock_quantity > 0)
    
    if search:
        conditions.append(
            Product.name.ilike(f"%{search}%") | 
            Product.description.ilike(f"%{search}%")
        )
    
    # 执行查询
    query = select(Product).where(*conditions)
    total_query = select(func.count()).select_from(query.subquery())
    
    # 分页
    query = query.offset(skip).limit(limit)
    
    # 异步执行查询
    result = await db.execute(query)
    products = result.scalars().all()
    
    total_result = await db.execute(total_query)
    total = total_result.scalar()
    
    return ProductListResponse(
        items=products,
        total=total,
        page=skip // limit + 1 if limit > 0 else 1,
        size=limit,
        pages=(total + limit - 1) // limit if limit > 0 else 1
    )

@router.post("/", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
async def create_product(
    *,
    db: AsyncSession = Depends(get_db),
    product_in: ProductCreate,
    current_user: User = Depends(get_current_active_superuser)
):
    """
    创建新商品（仅管理员）
    """
    # 检查 SKU 是否已存在
    existing_product = await ProductService.get_by_sku(db, product_in.sku)
    if existing_product:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="SKU 已存在"
        )
    
    product = await ProductService.create(db, product_in)
    return product

@router.put("/{product_id}", response_model=ProductResponse)
async def update_product(
    *,
    db: AsyncSession = Depends(get_db),
    product_id: int,
    product_in: ProductUpdate,
    current_user: User = Depends(get_current_active_superuser)
):
    """
    更新商品信息（仅管理员）
    """
    product = await ProductService.get(db, product_id)
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="商品不存在"
        )
    
    # 如果更新 SKU，检查是否重复
    if product_in.sku and product_in.sku != product.sku:
        existing = await ProductService.get_by_sku(db, product_in.sku)
        if existing:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="SKU 已存在"
            )
    
    product = await ProductService.update(db, product, product_in)
    return product

@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(
    *,
    db: AsyncSession = Depends(get_db),
    product_id: int
):
    """
    获取商品详情
    """
    product = await ProductService.get(db, product_id)
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="商品不存在"
        )
    
    return product
```

### 2. 订单处理服务

```python
# app/services/order_service.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import List, Optional
from decimal import Decimal

from app.models.order import Order, OrderStatus, OrderItem
from app.models.product import Product
from app.schemas.order import OrderCreate, OrderUpdate
from app.core.exceptions import BusinessException

class OrderService:
    """订单服务类"""
    
    @staticmethod
    async def create_order(
        db: AsyncSession, 
        order_in: OrderCreate, 
        user_id: int
    ) -> Order:
        """
        创建订单
        - 验证商品库存
        - 计算总金额
        - 创建订单记录
        """
        total_amount = Decimal('0')
        order_items = []
        
        # 验证每个订单项
        for item in order_in.items:
            # 获取商品信息
            result = await db.execute(
                select(Product).where(
                    Product.id == item.product_id,
                    Product.is_active == True
                )
            )
            product = result.scalar_one_or_none()
            
            if not product:
                raise BusinessException(f"商品 {item.product_id} 不存在或已下架")
            
            if product.stock_quantity < item.quantity:
                raise BusinessException(
                    f"商品 {product.name} 库存不足，"
                    f"剩余 {product.stock_quantity} 件"
                )
            
            # 计算金额
            item_price = product.price * item.quantity
            total_amount += item_price
            
            # 创建订单项
            order_item = OrderItem(
                product_id=item.product_id,
                quantity=item.quantity,
                unit_price=product.price,
                total_price=item_price
            )
            order_items.append(order_item)
            
            # 减少库存（悲观锁）
            product.stock_quantity -= item.quantity
        
        # 创建订单
        order = Order(
            user_id=user_id,
            total_amount=total_amount,
            shipping_address=order_in.shipping_address,
            status=OrderStatus.PENDING,
            items=order_items
        )
        
        db.add(order)
        await db.commit()
        await db.refresh(order)
        
        return order
    
    @staticmethod
    async def cancel_order(db: AsyncSession, order_id: int, user_id: int) -> Order:
        """
        取消订单
        - 只有待处理订单可以取消
        - 恢复商品库存
        """
        result = await db.execute(
            select(Order).where(
                Order.id == order_id,
                Order.user_id == user_id
            )
        )
        order = result.scalar_one_or_none()
        
        if not order:
            raise BusinessException("订单不存在")
        
        if order.status != OrderStatus.PENDING:
            raise BusinessException("只有待处理订单可以取消")
        
        # 恢复库存
        for item in order.items:
            result = await db.execute(
                select(Product).where(Product.id == item.product_id)
            )
            product = result.scalar_one()
            product.stock_quantity += item.quantity
        
        # 更新订单状态
        order.status = OrderStatus.CANCELLED
        await db.commit()
        await db.refresh(order)
        
        return order
```

### 3. 缓存优化策略

```python
# app/core/cache.py
from typing import Any, Optional
import