22.4 购物车与订单系统

FastAPI 综合项目实战：构建电子商务购物车与订单系统 API

FastAPI 教程

本教程详细指导如何使用 FastAPI 创建一个完整的电子商务 API，实现购物车管理和订单处理功能。适合初学者快速掌握 FastAPI 实战开发，提升 API 构建技能。

综合项目实战 - 电子商务 API 购物车与订单系统

介绍

欢迎来到 FastAPI 综合项目实战！在本教程中，我们将一起构建一个简单的电子商务系统 API，重点实现购物车管理和订单处理功能。这个项目非常适合初学者，通过实战加深对 FastAPI 的理解，学习如何设计 RESTful API、处理数据模型和集成数据库。

学习目标：

理解 FastAPI 的基本概念，如路由、依赖项和 Pydantic 模型。
实现购物车的 CRUD（创建、读取、更新、删除）操作。
设计订单系统，包括创建、查看和更新订单。
集成数据库以持久化存储数据。
可选添加身份验证来保护 API。

前置知识

在开始之前，确保你具备以下基础知识：

Python 基础：熟悉 Python 语法和基本数据类型。
HTTP 和 RESTful API：了解 HTTP 方法（GET、POST、PUT、DELETE）和 API 设计原则。
FastAPI 入门：建议先学习 FastAPI 的官方教程，了解路由和 Pydantic 模型的使用。
数据库概念：对关系型数据库（如 SQLite 或 PostgreSQL）有基本了解，但不是必须的，因为本教程会指导设置。

如果你是新用户，别担心，我会尽量用简单语言解释每一步。

环境设置

首先，我们需要设置开发环境。请按照以下步骤操作：

安装 Python：确保你的系统安装了 Python 3.7 或更高版本。你可以从 python.org 下载。
创建虚拟环境：打开终端或命令提示符，运行以下命令来创建一个虚拟环境（这有助于隔离项目依赖）：
```
python -m venv venv
source venv/bin/activate  # 在 Linux/Mac 上
venv\Scripts\activate    # 在 Windows 上
```
安装 FastAPI 和相关库：在虚拟环境中，运行以下命令安装 FastAPI、Uvicorn（用于运行服务器）和数据库库。我们使用 SQLAlchemy 作为 ORM 和 SQLite 作为数据库，以简化设置。
```
pip install fastapi uvicorn sqlalchemy pydantic
```
这安装了：
- fastapi：用于构建 API。
- uvicorn：ASGI 服务器，用于运行 FastAPI 应用。
- sqlalchemy：数据库 ORM，用于操作数据库。
- pydantic：用于数据验证和模型定义。
创建项目目录：在本地文件夹中，创建一个新目录作为项目根目录，并添加以下文件：
- main.py：主应用文件。
- models.py：定义数据库模型。
- schemas.py：定义 Pydantic 模型用于请求和响应。
- database.py：数据库连接配置。
- routers/：文件夹，用于存放购物车和订单的路由文件。

现在，环境已经准备就绪！

项目结构

让我们先规划项目结构，这样便于维护：

my_ecommerce_api/
│
├── main.py              # FastAPI 应用主入口
├── models.py            # SQLAlchemy 数据库模型
├── schemas.py           # Pydantic 模式用于验证
├── database.py          # 数据库连接和设置
├── routers/             # 路由模块文件夹
│   ├── cart.py          # 购物车相关路由
│   └── order.py         # 订单相关路由
└── requirements.txt     # 依赖列表（可选）

在 main.py 中，我们将初始化 FastAPI 应用并包含路由。

购物车 API 实现

模型定义

首先，在 models.py 中定义数据库模型。购物车通常包括用户、商品和数量。为了简化，我们假设每个用户有一个购物车，每个购物车可以包含多个商品项。

# models.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from database import Base  # 假设我们在 database.py 中定义了 Base

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    cart = relationship('Cart', back_populates='user', uselist=False)

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Integer)  # 假设价格以整数表示，单位为分

class Cart(Base):
    __tablename__ = 'carts'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship('User', back_populates='cart')
    items = relationship('CartItem', back_populates='cart')

class CartItem(Base):
    __tablename__ = 'cart_items'
    id = Column(Integer, primary_key=True, index=True)
    cart_id = Column(Integer, ForeignKey('carts.id'))
    product_id = Column(Integer, ForeignKey('products.id'))
    quantity = Column(Integer, default=1)
    cart = relationship('Cart', back_populates='items')
    product = relationship('Product')

在 schemas.py 中，使用 Pydantic 定义请求和响应模型，以验证输入数据和格式化输出。

# schemas.py
from pydantic import BaseModel
from typing import List, Optional

class CartItemBase(BaseModel):
    product_id: int
    quantity: int = 1

class CartItemCreate(CartItemBase):
    pass

class CartItemResponse(CartItemBase):
    id: int
    product_name: str  # 假设从产品模型获取
    class Config:
        orm_mode = True

class CartBase(BaseModel):
    user_id: int

class CartResponse(CartBase):
    id: int
    items: List[CartItemResponse] = []
    class Config:
        orm_mode = True

路由实现

接下来，在 routers/cart.py 中实现购物车相关的路由。我们将创建 CRUD 操作。

# routers/cart.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db  # 假设 database.py 中有获取数据库会话的函数
from models import Cart, CartItem, Product
from schemas import CartItemCreate, CartResponse, CartItemResponse

router = APIRouter(prefix='/cart', tags=['cart'])

@router.get('/', response_model=CartResponse)
async def get_cart(user_id: int, db: Session = Depends(get_db)):
    # 假设用户 ID 通过查询参数传递（在实际应用中，可能来自认证）
    cart = db.query(Cart).filter(Cart.user_id == user_id).first()
    if not cart:
        raise HTTPException(status_code=404, detail='Cart not found')
    return cart

@router.post('/items/', response_model=CartItemResponse)
async def add_item_to_cart(item: CartItemCreate, user_id: int, db: Session = Depends(get_db)):
    # 检查用户购物车是否存在，如果不存在则创建
    cart = db.query(Cart).filter(Cart.user_id == user_id).first()
    if not cart:
        cart = Cart(user_id=user_id)
        db.add(cart)
        db.commit()
        db.refresh(cart)
    
    # 检查产品是否存在
    product = db.query(Product).filter(Product.id == item.product_id).first()
    if not product:
        raise HTTPException(status_code=404, detail='Product not found')
    
    # 添加商品到购物车
    cart_item = CartItem(cart_id=cart.id, product_id=item.product_id, quantity=item.quantity)
    db.add(cart_item)
    db.commit()
    db.refresh(cart_item)
    # 返回响应，包含产品名称（假设从产品模型获取）
    return CartItemResponse(id=cart_item.id, product_id=cart_item.product_id, quantity=cart_item.quantity, product_name=product.name)

@router.put('/items/{item_id}', response_model=CartItemResponse)
async def update_item_quantity(item_id: int, quantity: int, db: Session = Depends(get_db)):
    cart_item = db.query(CartItem).filter(CartItem.id == item_id).first()
    if not cart_item:
        raise HTTPException(status_code=404, detail='Item not found in cart')
    cart_item.quantity = quantity
    db.commit()
    db.refresh(cart_item)
    product = db.query(Product).filter(Product.id == cart_item.product_id).first()
    return CartItemResponse(id=cart_item.id, product_id=cart_item.product_id, quantity=cart_item.quantity, product_name=product.name)

@router.delete('/items/{item_id}')
async def remove_item_from_cart(item_id: int, db: Session = Depends(get_db)):
    cart_item = db.query(CartItem).filter(CartItem.id == item_id).first()
    if not cart_item:
        raise HTTPException(status_code=404, detail='Item not found in cart')
    db.delete(cart_item)
    db.commit()
    return {'message': 'Item removed from cart'}

这实现了购物车的基本功能：获取购物车、添加商品、更新数量和移除商品。在实际应用中，你可能需要添加错误处理和更复杂的逻辑，比如检查库存。

订单系统 API 实现

模型定义

在 models.py 中添加订单相关模型。订单通常包括用户、订单项和状态。

# 在 models.py 中添加
class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    status = Column(String, default='pending')  # 状态如 pending, completed, cancelled
    items = relationship('OrderItem', back_populates='order')

class OrderItem(Base):
    __tablename__ = 'order_items'
    id = Column(Integer, primary_key=True, index=True)
    order_id = Column(Integer, ForeignKey('orders.id'))
    product_id = Column(Integer, ForeignKey('products.id'))
    quantity = Column(Integer)
    price = Column(Integer)  # 记录购买时的价格
    order = relationship('Order', back_populates='items')
    product = relationship('Product')

在 schemas.py 中添加订单相关的 Pydantic 模型。

# 在 schemas.py 中添加
class OrderItemBase(BaseModel):
    product_id: int
    quantity: int
    price: int  # 价格可以从产品获取，但记录以防变化

class OrderItemResponse(OrderItemBase):
    id: int
    product_name: str
    class Config:
        orm_mode = True

class OrderBase(BaseModel):
    user_id: int
    status: str = 'pending'

class OrderCreate(OrderBase):
    items: List[OrderItemBase]

class OrderResponse(OrderBase):
    id: int
    items: List[OrderItemResponse] = []
    class Config:
        orm_mode = True

路由实现

在 routers/order.py 中实现订单相关的路由。

# routers/order.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Order, OrderItem, Product, Cart, CartItem
from schemas import OrderCreate, OrderResponse

router = APIRouter(prefix='/orders', tags=['orders'])

@router.post('/', response_model=OrderResponse)
async def create_order(user_id: int, db: Session = Depends(get_db)):
    # 假设从购物车创建订单
    cart = db.query(Cart).filter(Cart.user_id == user_id).first()
    if not cart or not cart.items:
        raise HTTPException(status_code=400, detail='Cart is empty')
    
    # 创建订单
    order = Order(user_id=user_id, status='pending')
    db.add(order)
    db.commit()
    db.refresh(order)
    
    # 添加订单项
    for cart_item in cart.items:
        product = db.query(Product).filter(Product.id == cart_item.product_id).first()
        if not product:
            raise HTTPException(status_code=404, detail='Product not found')
        order_item = OrderItem(order_id=order.id, product_id=cart_item.product_id, 
                               quantity=cart_item.quantity, price=product.price)
        db.add(order_item)
    
    # 清空购物车（可选）
    db.query(CartItem).filter(CartItem.cart_id == cart.id).delete()
    db.commit()
    
    return order

@router.get('/', response_model=List[OrderResponse])
async def get_orders(user_id: int, db: Session = Depends(get_db)):
    orders = db.query(Order).filter(Order.user_id == user_id).all()
    return orders

@router.get('/{order_id}', response_model=OrderResponse)
async def get_order(order_id: int, db: Session = Depends(get_db)):
    order = db.query(Order).filter(Order.id == order_id).first()
    if not order:
        raise HTTPException(status_code=404, detail='Order not found')
    return order

@router.put('/{order_id}/status')
async def update_order_status(order_id: int, status: str, db: Session = Depends(get_db)):
    order = db.query(Order).filter(Order.id == order_id).first()
    if not order:
        raise HTTPException(status_code=404, detail='Order not found')
    order.status = status
    db.commit()
    return {'message': 'Order status updated'}

@router.delete('/{order_id}')
async def cancel_order(order_id: int, db: Session = Depends(get_db)):
    order = db.query(Order).filter(Order.id == order_id).first()
    if not order:
        raise HTTPException(status_code=404, detail='Order not found')
    db.delete(order)
    db.commit()
    return {'message': 'Order cancelled'}

这实现了订单的创建、查看、更新状态和取消功能。在实际应用中，你可能需要添加更多验证，比如检查用户权限或库存。

数据库集成

在 database.py 中设置数据库连接和会话。

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = 'sqlite:///./ecommerce.db'  # 使用 SQLite 数据库文件

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False} if SQLALCHEMY_DATABASE_URL.startswith('sqlite') else {})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

在 main.py 中初始化数据库并包含路由。

# main.py
from fastapi import FastAPI
from database import engine, Base
from routers import cart, order

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI(title='E-commerce API', description='A simple e-commerce API with cart and order system')

# 包含路由
app.include_router(cart.router)
app.include_router(order.router)

@app.get('/')
async def root():
    return {'message': 'Welcome to the E-commerce API!'}

身份验证与授权（可选）

为了安全，你可以添加身份验证。这里简要介绍使用 JWT（JSON Web Token）的简单方式。

安装额外的库：pip install python-jose[cryptography] passlib[bcrypt]
在 main.py 中添加依赖项来处理认证，例如使用 OAuth2 密码流。

由于本教程针对初学者，这部分可选；你可以参考 FastAPI 官方文档的“Security”部分来添加。

测试与部署

测试

使用 pytest 编写测试用例。创建一个 test_main.py 文件，测试 API 端点。例如：

# test_main.py
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() == {'message': 'Welcome to the E-commerce API!'}

运行测试：pytest

部署

本地运行：使用 Uvicorn 运行服务器：uvicorn main:app --reload，然后访问 http://127.0.0.1:8000/docs 查看自动生成的 API 文档。
部署到云：你可以将代码推送到 GitHub，然后使用云服务如 Heroku、AWS 或 Docker 部署。FastAPI 支持 ASGI，容易部署到各种平台。

总结

恭喜！你已经成功构建了一个基础的电子商务 API，包括购物车管理和订单系统。通过这个项目，你学习了：

如何使用 FastAPI 设计 RESTful API。
集成数据库和 ORM（SQLAlchemy）来存储数据。
实现购物车和订单的 CRUD 操作。
使用 Pydantic 进行数据验证和序列化。

进一步学习建议

添加用户认证和授权，保护 API 端点。
实现更复杂的业务逻辑，如库存管理、支付集成。
使用异步数据库操作提高性能。
部署应用到生产环境并添加监控。

希望这个教程对你有所帮助！如果有任何问题，请参考 FastAPI 官方文档或社区支持。祝你学习愉快！

注意：本教程代码为示例，可能需要根据实际需求调整。在实际应用中，请考虑错误处理、安全性和性能优化。

下一章 22.5 支付集成与Webhook

FastAPI 教程

22.4 购物车与订单系统

综合项目实战 - 电子商务 API 购物车与订单系统

介绍

前置知识

环境设置

项目结构

购物车 API 实现

模型定义

路由实现

订单系统 API 实现

模型定义

路由实现

数据库集成

身份验证与授权（可选）

测试与部署

测试

部署

总结

进一步学习建议

相关文档

Python 教程

Django 6中文教程

Flask 中文教程

NumPy 中文教程

Scikit-learn 中文教程

TensorFlow 中文手册