项目4: API开发¶

难度: ⭐⭐⭐⭐ 中高级时间: 4-6小时 涉及知识: FastAPI, Pydantic, 异步编程, API设计

🎯 项目目标¶

创建一个RESTful API服务： 1. 使用FastAPI框架 2. 实现CRUD操作 3. 数据验证和序列化 4. 自动API文档 5. 部署准备

📋 需求¶

功能需求¶

Bash

# 启动服务
uvicorn main:app --reload

# API端点
GET    /items          # 获取所有项目
GET    /items/{id}     # 获取单个项目
POST   /items          # 创建项目
PUT    /items/{id}     # 更新项目
DELETE /items/{id}     # 删除项目

# 功能
- 请求/响应数据验证
- 自动生成的API文档 (/docs)
- 错误处理
- 日志记录

技术要求¶

使用FastAPI框架
使用Pydantic进行数据验证
使用Uvicorn作为ASGI服务器
使用SQLAlchemy（可选，用于数据库）

🚀 实现步骤¶

步骤1: 环境准备¶

Bash

pip install fastapi uvicorn pydantic
# 可选：数据库支持
pip install sqlalchemy databases

步骤2: 基础API¶

Python

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict  # BaseModel是Pydantic的基类，提供数据验证和序列化
import uvicorn

app = FastAPI(
    title="Item API",
    description="A simple item management API",
    version="1.0.0"
)

# 数据模型
class Item(BaseModel):
    id: int
    name: str
    description: str | None = None
    price: float
    is_available: bool = True

    model_config = ConfigDict(json_schema_extra={
        "example": {
            "id": 1,
            "name": "iPhone",
            "description": "A smartphone",
            "price": 999.99,
            "is_available": True
        }
    })

class ItemCreate(BaseModel):
    name: str
    description: str | None = None
    price: float
    is_available: bool = True

class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
    is_available: bool | None = None

# 模拟数据库
items_db = {}

@app.get("/")  # @app.get/post定义HTTP路由及请求方法
def read_root():
    return {"message": "Welcome to Item API", "docs": "/docs"}

@app.get("/items", response_model=list[Item])
def get_items(skip: int = 0, limit: int = 10):
    """获取所有项目"""
    items = list(items_db.values())
    return items[skip : skip + limit]

@app.get("/items/{item_id}", response_model=Item)
def get_item(item_id: int):
    """获取单个项目"""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

@app.post("/items", response_model=Item, status_code=201)
def create_item(item: ItemCreate):
    """创建新项目"""
    item_id = max(items_db.keys(), default=0) + 1
    new_item = Item(id=item_id, **item.model_dump())  # model_dump()将Pydantic模型转为字典
    items_db[item_id] = new_item
    return new_item

@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: int, item_update: ItemUpdate):
    """更新项目"""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")

    stored_item = items_db[item_id]
    update_data = item_update.model_dump(exclude_unset=True)

    for field, value in update_data.items():
        setattr(stored_item, field, value)  # setattr()动态设置对象属性

    items_db[item_id] = stored_item
    return stored_item

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    """删除项目"""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")

    del items_db[item_id]
    return {"message": "Item deleted successfully"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

步骤3: 添加数据库支持¶

Python

# database.py
from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import sessionmaker

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

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

class Base(DeclarativeBase):
    pass

class ItemModel(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)
    price = Column(Float)
    is_available = Column(Boolean, default=True)

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

# 依赖注入
def get_db():
    db = SessionLocal()
    try:
        yield db  # yield将函数变为生成器，这野yield前后分别为资源初始化和清理逻辑
    finally:
        db.close()

Python

# main_with_db.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
import uvicorn

from database import get_db, ItemModel
from schemas import Item, ItemCreate, ItemUpdate

app = FastAPI(title="Item API with Database")

@app.post("/items", response_model=Item)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):  # Depends()声明依赖注入，自动解析和注入依赖
    db_item = ItemModel(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items", response_model=list[Item])
def get_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    items = db.query(ItemModel).offset(skip).limit(limit).all()
    return items

@app.get("/items/{item_id}", response_model=Item)
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(ItemModel).filter(ItemModel.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: int, item_update: ItemUpdate, db: Session = Depends(get_db)):
    item = db.query(ItemModel).filter(ItemModel.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")

    for field, value in item_update.model_dump(exclude_unset=True).items():
        setattr(item, field, value)

    db.commit()
    db.refresh(item)
    return item

@app.delete("/items/{item_id}")
def delete_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(ItemModel).filter(ItemModel.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")

    db.delete(item)
    db.commit()
    return {"message": "Item deleted"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

步骤4: 高级功能¶

Python

# advanced_api.py
from fastapi import FastAPI, HTTPException, Query, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Advanced Item API",
    version="2.0.0"
)

# CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 请求日志中间件
@app.middleware("http")
async def log_requests(request, call_next):  # async def定义协程函数，await暂停等待异步结果
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time
    logger.info(f"{request.method} {request.url.path} - {response.status_code} - {duration:.2f}s")
    return response

# 分页响应（需要前面步骤2中定义的Item模型）
class PaginatedResponse(BaseModel):
    items: list[Item]
    total: int
    skip: int
    limit: int

@app.get("/items", response_model=PaginatedResponse)
def get_items_paginated(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    search: str | None = None,
    min_price: float | None = None,
    max_price: float | None = None
):
    """获取项目列表（支持分页和筛选）"""
    items = list(items_db.values())

    # 搜索过滤
    if search:
        items = [i for i in items if search.lower() in i.name.lower()]

    # 价格过滤
    if min_price is not None:
        items = [i for i in items if i.price >= min_price]
    if max_price is not None:
        items = [i for i in items if i.price <= max_price]

    total = len(items)
    items = items[skip : skip + limit]

    return PaginatedResponse(
        items=items,
        total=total,
        skip=skip,
        limit=limit
    )

📝 扩展挑战¶

用户认证 - 添加JWT token认证
速率限制 - 使用slowapi限制请求频率
缓存 - 使用Redis缓存响应
后台任务 - 使用Celery处理耗时操作
测试 - 使用pytest和TestClient编写测试
部署 - 使用Docker容器化部署

🎯 完成标准¶

实现完整的CRUD操作
使用Pydantic进行数据验证
自动生成API文档可访问
有适当的错误处理
代码结构清晰，模块化
可选：添加数据库支持

💡 提示¶

使用/docs查看自动生成的Swagger UI
使用/redoc查看ReDoc文档
利用Pydantic的验证功能减少手动检查
使用依赖注入管理共享资源
异步函数使用async def提升性能

📚 参考资源¶

🚀 下一步¶

完成后，尝试： - 项目5: 完整ML项目 - 将之前的ML模型封装成API - 学习微服务架构

记住: 好的API设计要考虑到使用者的体验！