综合项目实战：FastAPI构建电子商务API商品目录与搜索

介绍

欢迎来到FastAPI实战教程！在这个项目中，我们将一起构建一个电子商务API，专注于商品目录管理和搜索功能。FastAPI是一个现代、快速（高性能）的Web框架，基于Python 3.6+的类型提示和标准库，易于学习和使用。通过这个项目，你将掌握如何用FastAPI开发RESTful API，并对电子商务场景有深入了解。

项目目标：

• 创建一个FastAPI应用，实现商品目录的增删改查（CRUD）操作。
• 集成搜索功能，允许用户根据关键词、价格范围等条件查询商品。
• 学习使用Pydantic进行数据验证和序列化。
• 实践数据库操作（这里假设使用SQLite或PostgreSQL，但教程保持通用性）。
• 编写测试以确保API的可靠性。

环境设置

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

mkdir ecommerce-api
cd ecommerce-api
python -m venv venv
# 激活虚拟环境：
# Windows: venv\Scripts\activate
# Mac/Linux: source venv/bin/activate

安装必要的依赖：

pip install fastapi uvicorn sqlalchemy pydantic

• FastAPI：Web框架。
• Uvicorn：ASGI服务器，用于运行FastAPI应用。
• SQLAlchemy：数据库ORM（对象关系映射），用于数据库操作。
• Pydantic：数据验证和设置管理。

可选地，如果你计划使用数据库，可以安装sqlite3（Python内置）或psycopg2（用于PostgreSQL）。

数据模型定义

使用Pydantic定义商品的数据模型。在项目根目录创建文件models.py。

from pydantic import BaseModel, Field
from typing import Optional, List

# 商品模型，定义API请求和响应的数据结构
class Product(BaseModel):
    id: Optional[int] = None  # 可选，因为数据库可能自动生成ID
    name: str = Field(..., min_length=1, max_length=100, description="商品名称")
    description: Optional[str] = Field(None, max_length=500, description="商品描述")
    price: float = Field(..., gt=0, description="商品价格，必须大于0")
    category: str = Field(..., min_length=1, max_length=50, description="商品类别")
    stock: int = Field(..., ge=0, description="库存数量，必须大于等于0")

    class Config:
        schema_extra = {
            "example": {
                "name": "智能手机",
                "description": "最新款智能手机，高性能处理器",
                "price": 2999.99,
                "category": "电子产品",
                "stock": 100
            }
        }

# 搜索查询模型，用于搜索端点
class ProductSearch(BaseModel):
    keyword: Optional[str] = Field(None, description="搜索关键词，如商品名称或描述")
    min_price: Optional[float] = Field(None, ge=0, description="最低价格")
    max_price: Optional[float] = Field(None, ge=0, description="最高价格")
    category: Optional[str] = Field(None, description="商品类别")

这个模型定义了商品的基本属性和搜索参数。Pydantic会自动进行数据验证，确保输入的有效性。

创建FastAPI应用和API端点

在项目根目录创建主文件main.py。

from fastapi import FastAPI, HTTPException, Depends, Query
from typing import List, Optional
from models import Product, ProductSearch

# 初始化FastAPI应用
app = FastAPI(title="电子商务API", version="1.0.0", description="一个简单的电子商务API，包含商品目录和搜索功能")

# 内存中的模拟数据库（实际项目中应使用数据库）
products_db = []

# API端点：获取所有商品
@app.get("/products", response_model=List[Product])
async def get_products():
    """返回所有商品列表。"""
    return products_db

# API端点：创建新商品
@app.post("/products", response_model=Product, status_code=201)
async def create_product(product: Product):
    """添加一个新商品到目录。"""
    # 模拟生成ID（实际应用中由数据库处理）
    product.id = len(products_db) + 1
    products_db.append(product)
    return product

# API端点：根据ID获取单个商品
@app.get("/products/{product_id}", response_model=Product)
async def get_product(product_id: int):
    """根据ID获取商品。"""
    for product in products_db:
        if product.id == product_id:
            return product
    raise HTTPException(status_code=404, detail="商品未找到")

# API端点：更新商品
@app.put("/products/{product_id}", response_model=Product)
async def update_product(product_id: int, updated_product: Product):
    """更新现有商品信息。"""
    for i, product in enumerate(products_db):
        if product.id == product_id:
            updated_product.id = product_id  # 保持ID不变
            products_db[i] = updated_product
            return updated_product
    raise HTTPException(status_code=404, detail="商品未找到")

# API端点：删除商品
@app.delete("/products/{product_id}", status_code=204)
async def delete_product(product_id: int):
    """从目录中删除商品。"""
    for i, product in enumerate(products_db):
        if product.id == product_id:
            products_db.pop(i)
            return
    raise HTTPException(status_code=404, detail="商品未找到")

# API端点：搜索商品
@app.get("/products/search/", response_model=List[Product])
async def search_products(
    keyword: Optional[str] = Query(None, description="搜索关键词"),
    min_price: Optional[float] = Query(None, ge=0, description="最低价格"),
    max_price: Optional[float] = Query(None, ge=0, description="最高价格"),
    category: Optional[str] = Query(None, description="商品类别")
):
    """根据关键词、价格范围和类别搜索商品。"""
    results = products_db
    if keyword:
        results = [p for p in results if keyword.lower() in p.name.lower() or (p.description and keyword.lower() in p.description.lower())]
    if min_price is not None:
        results = [p for p in results if p.price >= min_price]
    if max_price is not None:
        results = [p for p in results if p.price <= max_price]
    if category:
        results = [p for p in results if category.lower() == p.category.lower()]
    return results

这个代码创建了一个基本的FastAPI应用，实现了商品目录的CRUD操作和搜索功能。为了简化，我们使用内存列表模拟数据库。实际项目中，你应该集成SQLAlchemy或类似ORM来连接真实数据库。

搜索功能实现详解

在上面的搜索端点中，我们使用查询参数来实现搜索。这只是一个简单示例，适合小型应用。对于大型电子商务网站，你可能需要：

• 使用数据库索引：确保搜索性能，例如在商品名称和描述字段上创建全文索引。
• 集成搜索引擎：如Elasticsearch或Whoosh，以支持更高级的全文搜索、模糊匹配等。
• 分页和排序：在搜索端点添加分页参数（如skip和limit）和排序选项。

为了扩展搜索功能，你可以修改search_products端点：

from fastapi import Query
from typing import Optional

@app.get("/products/search/", response_model=List[Product])
async def search_products(
    keyword: Optional[str] = Query(None),
    min_price: Optional[float] = Query(None, ge=0),
    max_price: Optional[float] = Query(None, ge=0),
    category: Optional[str] = Query(None),
    skip: int = Query(0, ge=0, description="跳过前N条结果"),
    limit: int = Query(10, le=100, description="返回结果数量，最大100")
):
    results = products_db
    # 应用过滤条件
    if keyword:
        results = [p for p in results if keyword.lower() in p.name.lower() or (p.description and keyword.lower() in p.description.lower())]
    if min_price is not None:
        results = [p for p in results if p.price >= min_price]
    if max_price is not None:
        results = [p for p in results if p.price <= max_price]
    if category:
        results = [p for p in results if category.lower() == p.category.lower()]
    # 分页处理
    paginated_results = results[skip:skip + limit]
    return paginated_results

测试API

FastAPI内置了测试支持，可以使用TestClient。创建一个测试文件test_api.py。

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_product():
    response = client.post("/products", json={"name": "测试商品", "description": "这是一个测试商品", "price": 100.0, "category": "测试", "stock": 10})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "测试商品"
    assert "id" in data

def test_search_products():
    # 先创建一个商品用于测试搜索
    client.post("/products", json={"name": "笔记本电脑", "price": 5000.0, "category": "电子产品", "stock": 5})
    response = client.get("/products/search/?keyword=笔记本")
    assert response.status_code == 200
    data = response.json()
    assert len(data) > 0
    assert data[0]["name"] == "笔记本电脑"

def test_get_product_not_found():
    response = client.get("/products/999")
    assert response.status_code == 404

运行测试：

pytest test_api.py

部署建议

一旦开发完成，你可以部署FastAPI应用到生产环境：

• 使用Uvicorn：运行uvicorn main:app --host 0.0.0.0 --port 8000启动服务器。
• 使用Gunicorn：对于多进程支持，结合Gunicorn和Uvicorn workers。
• 容器化：使用Docker构建镜像，便于部署到云平台如AWS、Google Cloud或Heroku。
• 数据库连接：在生产环境中替换内存数据库为PostgreSQL或MySQL，并设置连接池。

总结

在这个项目中，我们从头构建了一个电子商务API，包括商品目录的管理（CRUD操作）和搜索功能。关键学习点包括：

• FastAPI的基本使用和API端点定义。
• Pydantic模型用于数据验证和序列化。
• 模拟数据库操作（在实际项目中应使用数据库ORM）。
• 实现搜索功能的基础和扩展方法。
• 编写测试以确保代码质量。

通过这个实战项目，你已经掌握了FastAPI的核心技能。建议下一步学习数据库集成（如使用SQLAlchemy）、身份验证和授权、以及高级搜索优化。祝你学习愉快，继续探索FastAPI的更多功能！

扩展资源

• FastAPI官方文档：https://fastapi.tiangolo.com/
• SQLAlchemy文档：https://www.sqlalchemy.org/
• 部署FastAPI指南：参考官方文档的部署部分。