综合项目实战：电子商务 API 库存管理与通知系统

简介

欢迎来到这个FastAPI实战教程！我们将一起构建一个完整的电子商务API库存管理与通知系统。这个项目将帮助你掌握FastAPI的核心概念，并通过实战提升开发技能。如果你是新手，不用担心，教程会从基础开始，确保简单易懂。

本项目实现以下功能：

• 库存管理：对产品库存进行增删改查操作。
• 通知系统：当库存低于预设阈值时，自动发送通知（如电子邮件）。

我们将使用FastAPI（一个现代、快速的Python Web框架）作为后端，结合SQLAlchemy进行数据库操作，Pydantic进行数据验证。

环境设置

首先，确保你已安装Python 3.8+。然后，创建一个新目录并设置虚拟环境。

mkdir ecommerce-inventory-api
cd ecommerce-inventory-api
python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate  # Windows

安装所需依赖：

pip install fastapi uvicorn sqlalchemy pydantic python-dotenv aiosmtplib

项目结构

创建一个清晰的项目结构，便于管理代码：

ecommerce-inventory-api/
│   main.py               # FastAPI应用主文件
│   .env                  # 环境变量文件
│   requirements.txt      # 依赖列表
├── models.py             # 数据库模型
├── schemas.py            # Pydantic数据验证模型
├── database.py           # 数据库连接配置
├── crud.py               # 数据库操作函数
├── routers/              # 路由模块
│   ├── __init__.py
│   ├── inventory.py      # 库存管理路由
│   └── notifications.py  # 通知路由
└── config.py             # 配置设置

数据库设计

在models.py中定义数据模型。我们将创建一个简单的库存表。

from sqlalchemy import Column, Integer, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Inventory(Base):
    __tablename__ = "inventory"
    
    id = Column(Integer, primary_key=True, index=True)
    product_name = Column(String, nullable=False)
    stock_quantity = Column(Integer, nullable=False, default=0)
    price = Column(Float, nullable=False)
    threshold = Column(Integer, nullable=False, default=10)  # 库存阈值
    created_at = Column(DateTime, default=datetime.utcnow)

数据库连接

在database.py中配置数据库连接。我们使用SQLite作为示例数据库。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./inventory.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

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

数据验证模型

在schemas.py中定义Pydantic模型，用于API请求和响应的数据验证。

from pydantic import BaseModel
from datetime import datetime

class InventoryBase(BaseModel):
    product_name: str
    stock_quantity: int
    price: float
    threshold: int = 10

class InventoryCreate(InventoryBase):
    pass

class InventoryUpdate(BaseModel):
    stock_quantity: int

class InventoryResponse(InventoryBase):
    id: int
    created_at: datetime
    
    class Config:
        orm_mode = True

创建FastAPI应用

在main.py中初始化FastAPI应用，并添加路由。

from fastapi import FastAPI, Depends
from database import get_db
from sqlalchemy.orm import Session
import routers.inventory as inventory_router
import routers.notifications as notifications_router

app = FastAPI(title="Ecommerce Inventory API", description="一个用于库存管理和通知的API系统")

# 包含路由
app.include_router(inventory_router.router, prefix="/api/v1/inventory", tags=["inventory"])
app.include_router(notifications_router.router, prefix="/api/v1/notifications", tags=["notifications"])

@app.get("/")
def read_root():
    return {"message": "欢迎使用电子商务库存管理与通知系统API"}

库存管理路由

在routers/inventory.py中实现库存管理的CRUD操作。

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Inventory
from schemas import InventoryCreate, InventoryResponse, InventoryUpdate
from crud import create_inventory, get_inventory, update_inventory, delete_inventory, check_threshold

router = APIRouter()

@router.post("/", response_model=InventoryResponse)
def create_new_inventory(inventory: InventoryCreate, db: Session = Depends(get_db)):
    return create_inventory(db=db, inventory=inventory)

@router.get("/{inventory_id}", response_model=InventoryResponse)
def read_inventory(inventory_id: int, db: Session = Depends(get_db)):
    db_inventory = get_inventory(db, inventory_id=inventory_id)
    if db_inventory is None:
        raise HTTPException(status_code=404, detail="库存项未找到")
    return db_inventory

@router.put("/{inventory_id}", response_model=InventoryResponse)
def update_existing_inventory(inventory_id: int, inventory_update: InventoryUpdate, db: Session = Depends(get_db)):
    db_inventory = update_inventory(db, inventory_id=inventory_id, inventory_update=inventory_update)
    if db_inventory is None:
        raise HTTPException(status_code=404, detail="库存项未找到")
    # 检查库存阈值并触发通知
    if check_threshold(db_inventory):
        # 这里可以调用通知逻辑，稍后实现
        pass
    return db_inventory

@router.delete("/{inventory_id}")
def delete_existing_inventory(inventory_id: int, db: Session = Depends(get_db)):
    success = delete_inventory(db, inventory_id=inventory_id)
    if not success:
        raise HTTPException(status_code=404, detail="库存项未找到")
    return {"message": "库存项已删除"}

在crud.py中定义数据库操作函数。

from sqlalchemy.orm import Session
from models import Inventory
from schemas import InventoryCreate, InventoryUpdate

def create_inventory(db: Session, inventory: InventoryCreate):
    db_inventory = Inventory(**inventory.dict())
    db.add(db_inventory)
    db.commit()
    db.refresh(db_inventory)
    return db_inventory

def get_inventory(db: Session, inventory_id: int):
    return db.query(Inventory).filter(Inventory.id == inventory_id).first()

def update_inventory(db: Session, inventory_id: int, inventory_update: InventoryUpdate):
    db_inventory = get_inventory(db, inventory_id)
    if db_inventory:
        db_inventory.stock_quantity = inventory_update.stock_quantity
        db.commit()
        db.refresh(db_inventory)
    return db_inventory

def delete_inventory(db: Session, inventory_id: int):
    db_inventory = get_inventory(db, inventory_id)
    if db_inventory:
        db.delete(db_inventory)
        db.commit()
        return True
    return False

def check_threshold(inventory: Inventory):
    return inventory.stock_quantity < inventory.threshold

通知系统集成

现在，实现通知功能。我们将使用SMTP发送电子邮件通知。首先，在.env文件中设置环境变量。

SMTP_SERVER=smtp.gmail.com
SMTP_PORT=587
SMTP_USERNAME=your-email@gmail.com
SMTP_PASSWORD=your-app-password

在routers/notifications.py中添加通知路由，并在config.py中加载配置。

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

SMTP_SERVER = os.getenv("SMTP_SERVER")
SMTP_PORT = int(os.getenv("SMTP_PORT", 587))
SMTP_USERNAME = os.getenv("SMTP_USERNAME")
SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")

# routers/notifications.py
from fastapi import APIRouter, BackgroundTasks
import aiosmtplib
from email.message import EmailMessage
from config import SMTP_SERVER, SMTP_PORT, SMTP_USERNAME, SMTP_PASSWORD

router = APIRouter()

async def send_email_notification(product_name: str, stock_quantity: int, threshold: int):
    msg = EmailMessage()
    msg["Subject"] = f"库存警报: {product_name} 库存低于阈值"
    msg["From"] = SMTP_USERNAME
    msg["To"] = "admin@example.com"  # 替换为实际接收邮箱
    msg.set_content(f"产品 {product_name} 的库存为 {stock_quantity}，低于阈值 {threshold}。请及时补货！")
    
    try:
        await aiosmtplib.send(msg, hostname=SMTP_SERVER, port=SMTP_PORT, username=SMTP_USERNAME, password=SMTP_PASSWORD, use_tls=True)
        print("通知邮件已发送")
    except Exception as e:
        print(f"发送邮件失败: {e}")

@router.post("/trigger")
async def trigger_notification(product_name: str, stock_quantity: int, threshold: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_notification, product_name, stock_quantity, threshold)
    return {"message": "通知已触发，将在后台发送"}

在crud.py中修改update_inventory函数以集成通知。

# 在update_inventory函数中添加通知触发
from routers.notifications import send_email_notification
from fastapi import BackgroundTasks

# 注意：为了简化，这里直接调用；实际项目中应使用依赖注入或事件系统。
# 在update路由中，可以传递background_tasks参数。

修改routers/inventory.py的update端点：

# 更新import
from fastapi import BackgroundTasks

@router.put("/{inventory_id}", response_model=InventoryResponse)
async def update_existing_inventory(inventory_id: int, inventory_update: InventoryUpdate, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    db_inventory = update_inventory(db, inventory_id=inventory_id, inventory_update=inventory_update)
    if db_inventory is None:
        raise HTTPException(status_code=404, detail="库存项未找到")
    if check_threshold(db_inventory):
        background_tasks.add_task(send_email_notification, db_inventory.product_name, db_inventory.stock_quantity, db_inventory.threshold)
    return db_inventory

测试与运行

编写简单的测试用例，然后运行应用。在项目根目录创建test_main.py。

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "欢迎使用电子商务库存管理与通知系统API"}

def test_create_inventory():
    response = client.post("/api/v1/inventory/", json={"product_name": "Test Product", "stock_quantity": 50, "price": 10.0})
    assert response.status_code == 200
    data = response.json()
    assert data["product_name"] == "Test Product"

运行测试：

pytest test_main.py

启动FastAPI应用：

uvicorn main:app --reload

访问API文档：http://127.0.0.1:8000/docs

部署与SEO优化建议

完成开发后，考虑部署到云平台如Heroku、AWS或使用Docker容器化。在部署时，设置环境变量确保安全。

对于SEO，优化您的API文档：

• 标题和描述：在FastAPI应用初始化时设置，如我们已在main.py中设置。
• 关键词：在内容中使用相关术语，如“库存管理”、“电子商务API”。
• 友好URL：使用清晰的端点结构，如/api/v1/inventory。

总结

通过这个实战项目，你学会了如何使用FastAPI构建一个功能完整的电子商务库存管理与通知系统。从环境设置到部署，涵盖了FastAPI的核心功能，适合新人学习和实践。继续探索FastAPI的高级特性，如依赖注入、WebSocket或安全认证，以扩展项目。

希望这个教程对你有帮助！如果有任何问题，欢迎查阅FastAPI官方文档或在线社区。