# 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