09-实战:Python Web API 开发

通过开发一个完整的 FastAPI Todo API,综合运用 Claude Code 的 MCP、Sub Agent、Hooks、Skills 等核心功能。


一、项目概述

1.1 项目信息

  • 项目名称: FastAPI Todo API
  • 项目地址: github.com/example/fastapi-todo-api
  • 技术栈: FastAPI + SQLAlchemy + Pydantic + PostgreSQL + pytest
  • 项目类型: RESTful Web API

1.2 功能特性

FastAPI Todo API - 任务管理 API

核心功能:
- 用户注册/登录 (JWT 认证)
- 任务的增删改查
- 任务分类管理
- 任务优先级设置
- 任务完成状态追踪

1.3 项目结构

fastapi-todo-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用入口
│   ├── config.py            # 配置管理
│   ├── models/              # SQLAlchemy 模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── todo.py
│   │   └── category.py
│   ├── schemas/             # Pydantic 模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── todo.py
│   │   └── category.py
│   ├── routers/             # API 路由
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── todos.py
│   │   └── categories.py
│   ├── services/            # 业务逻辑
│   │   ├── __init__.py
│   │   ├── auth_service.py
│   │   ├── todo_service.py
│   │   └── category_service.py
│   └── utils/               # 工具函数
│       ├── __init__.py
│       ├── security.py
│       └── validators.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py          # pytest fixtures
│   ├── test_auth.py
│   ├── test_todos.py
│   └── test_categories.py
├── alembic/                 # 数据库迁移
├── requirements.txt
├── requirements-dev.txt
├── docker-compose.yml
├── Dockerfile
├── CLAUDE.md               # Claude Code 配置
├── .env.example
├── .gitignore
└── README.md

二、环境准备

2.1 初始化项目

在 Claude Code 中执行:

# 创建项目目录
mkdir fastapi-todo-api
cd fastapi-todo-api

# 初始化项目
> 帮我初始化一个 FastAPI 项目,包含:
  1. 创建上述目录结构
  2. 创建基础配置文件
  3. 创建 requirements.txt
  4. 创建 README.md
  5. 初始化 Git 仓库
  6. 创建 .gitignore

Claude 会自动创建完整的项目结构。

2.2 配置文件

requirements.txt:

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
asyncpg==0.29.0
alembic==1.12.1
pydantic==2.5.0
pydantic-settings==2.1.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
pytest==7.4.3
pytest-asyncio==0.21.1
httpx==0.25.2

requirements-dev.txt:

-r requirements.txt
black==23.11.0
flake8==6.1.0
mypy==1.7.1
isort==5.12.0
pytest-cov==4.1.0

CLAUDE.md:

# CLAUDE.md - FastAPI Todo API 项目配置

## 项目信息

- **名称**: FastAPI Todo API
- **技术栈**: FastAPI + SQLAlchemy 2.0 + PostgreSQL
- **Python版本**: 3.9+

## 开发规范

### 代码风格
- 使用 Black 格式化(行宽 100)
- 使用 Flake8 检查(max-line-length=100)
- 所有函数必须添加类型注解
- 使用 Google Style Docstrings

### 架构规范
- 使用 SQLAlchemy 2.0 新语法(select(), session.get())
- 使用 Pydantic V2 BaseModel
- 依赖注入管理数据库会话
- 统一错误处理格式

### API 规范
- 使用 APIRouter 组织路由
- 响应模型使用 Pydantic
- 统一错误响应格式:{"detail": "错误消息"}
- 认证使用 JWT Bearer Token

## 常用命令

```bash
# 启动开发服务器
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 运行测试
pytest tests/ -v --cov=app --cov-report=term-missing

# 数据库迁移
alembic revision --autogenerate -m "描述"
alembic upgrade head

# 代码格式化
black app/ tests/
isort app/ tests/

# 类型检查
mypy app/

项目结构

app/
├── main.py          # FastAPI 应用
├── config.py        # 配置
├── models/          # SQLAlchemy 模型
├── schemas/         # Pydantic 模型
├── routers/         # API 路由
├── services/        # 业务逻辑
└── utils/           # 工具函数

### 2.3 配置 MCP 服务器

```bash
> 配置 MCP 服务器:
  1. 添加 Browser MCP 用于测试 API
  2. 添加 Context7 MCP 用于查询文档
  3. 配置 File System MCP 访问项目文件

三、使用 Claude Code 开发

3.1 创建数据模型

使用 Sub Agent 并行创建:

> 请并行创建以下 SQLAlchemy 模型:

  任务1 - User 模型:
  创建 app/models/user.py:
  - 字段:id, username, email, hashed_password, is_active, created_at
  - 关系:一个用户有多个 Todos
  - 使用 SQLAlchemy 2.0 语法

  任务2 - Todo 模型:
  创建 app/models/todo.py:
  - 字段:id, title, description, completed, priority, due_date, 
         created_at, updated_at, user_id, category_id
  - 关系:属于 User,属于 Category
  - 优先级:low, medium, high, urgent

  任务3 - Category 模型:
  创建 app/models/category.py:
  - 字段:id, name, description, user_id
  - 关系:属于 User,有多个 Todos

  所有模型使用 SQLAlchemy 2.0 的 DeclarativeBase,
  使用 Mapped[] 类型注解语法。

生成的模型代码示例:

# app/models/user.py
from datetime import datetime
from typing import List, Optional

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base


class User(Base):
    """用户模型"""
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    
    # 关系
    todos: Mapped[List["Todo"]] = relationship(back_populates="user")
    categories: Mapped[List["Category"]] = relationship(back_populates="user")

3.2 创建 Pydantic Schemas

> 为所有模型创建 Pydantic schemas:
  1. UserCreate, UserResponse, UserLogin
  2. TodoCreate, TodoUpdate, TodoResponse, TodoList
  3. CategoryCreate, CategoryResponse
  
  使用 Pydantic V2 语法,所有字段添加 Field 约束。

示例:

# app/schemas/todo.py
from datetime import date, datetime
from typing import Optional

from pydantic import BaseModel, Field


class TodoBase(BaseModel):
    """Todo 基础模型"""
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=1000)
    priority: str = Field(default="medium", pattern="^(low|medium|high|urgent)$")
    due_date: Optional[date] = None


class TodoCreate(TodoBase):
    """创建 Todo 请求"""
    category_id: Optional[int] = None


class TodoUpdate(BaseModel):
    """更新 Todo 请求"""
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=1000)
    completed: Optional[bool] = None
    priority: Optional[str] = Field(None, pattern="^(low|medium|high|urgent)$")
    due_date: Optional[date] = None
    category_id: Optional[int] = None


class TodoResponse(TodoBase):
    """Todo 响应模型"""
    id: int
    completed: bool
    created_at: datetime
    updated_at: datetime
    user_id: int
    category_id: Optional[int]
    
    class Config:
        from_attributes = True

3.3 创建 API 路由

使用 Skills 规范生成:

> 创建 API 路由文件:
  1. app/routers/auth.py - 认证路由(登录、注册)
  2. app/routers/todos.py - Todo CRUD
  3. app/routers/categories.py - 分类管理
  
  使用 fastapi-api Skill 的规范:
  - 使用 APIRouter
  - 使用依赖注入
  - 添加适当的 tags
  - 实现完整的错误处理

示例路由:

# app/routers/todos.py
from typing import List

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.models.user import User
from app.schemas.todo import TodoCreate, TodoResponse, TodoUpdate
from app.services.auth import get_current_user
from app.services.todo_service import TodoService

router = APIRouter(prefix="/todos", tags=["todos"])


@router.get("/", response_model=List[TodoResponse])
async def list_todos(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
) -> List[TodoResponse]:
    """获取当前用户的所有 Todo"""
    service = TodoService(db)
    todos = await service.get_user_todos(current_user.id, skip, limit)
    return todos


@router.post("/", response_model=TodoResponse, status_code=status.HTTP_201_CREATED)
async def create_todo(
    todo: TodoCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
) -> TodoResponse:
    """创建新 Todo"""
    service = TodoService(db)
    return await service.create_todo(todo, current_user.id)


@router.get("/{todo_id}", response_model=TodoResponse)
async def get_todo(
    todo_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
) -> TodoResponse:
    """获取指定 Todo"""
    service = TodoService(db)
    todo = await service.get_todo(todo_id, current_user.id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Todo not found"
        )
    return todo


@router.put("/{todo_id}", response_model=TodoResponse)
async def update_todo(
    todo_id: int,
    todo_update: TodoUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
) -> TodoResponse:
    """更新 Todo"""
    service = TodoService(db)
    todo = await service.update_todo(todo_id, current_user.id, todo_update)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Todo not found"
        )
    return todo


@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(
    todo_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
) -> None:
    """删除 Todo"""
    service = TodoService(db)
    success = await service.delete_todo(todo_id, current_user.id)
    if not success:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Todo not found"
        )

3.4 创建业务逻辑层

> 创建 Service 层:
  1. TodoService - 处理 Todo 业务逻辑
  2. CategoryService - 处理分类业务逻辑
  3. AuthService - 处理认证业务逻辑
  
  所有服务类使用依赖注入获取数据库会话。

3.5 自动化测试

使用自动化 Hooks 生成测试:

> 配置自动化测试流程:
  
  创建 tests/ 下的测试文件:
  1. conftest.py - 配置 fixtures
  2. test_auth.py - 认证测试
  3. test_todos.py - Todo API 测试
  4. test_categories.py - 分类 API 测试
  
  然后运行测试验证。

Claude 会自动:
1. 生成测试代码
2. 运行 pytest
3. 显示测试结果

示例测试:

# tests/test_todos.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from app.main import app
from app.models.todo import Todo
from app.models.user import User


@pytest.fixture
async def auth_headers(async_client: AsyncClient, test_user: User) -> dict:
    """获取认证 headers"""
    response = await async_client.post(
        "/api/v1/auth/login",
        json={"username": test_user.username, "password": "testpassword"}
    )
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


@pytest.mark.asyncio
async def test_create_todo(
    async_client: AsyncClient,
    auth_headers: dict,
    db_session: AsyncSession
):
    """测试创建 Todo"""
    response = await async_client.post(
        "/api/v1/todos/",
        headers=auth_headers,
        json={
            "title": "Test Todo",
            "description": "Test description",
            "priority": "high"
        }
    )
    
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Test Todo"
    assert data["priority"] == "high"
    assert data["completed"] is False


@pytest.mark.asyncio
async def test_list_todos(
    async_client: AsyncClient,
    auth_headers: dict,
    test_user: User,
    db_session: AsyncSession
):
    """测试获取 Todo 列表"""
    # 创建测试数据
    todo1 = Todo(title="Todo 1", user_id=test_user.id)
    todo2 = Todo(title="Todo 2", user_id=test_user.id)
    db_session.add_all([todo1, todo2])
    await db_session.commit()
    
    response = await async_client.get(
        "/api/v1/todos/",
        headers=auth_headers
    )
    
    assert response.status_code == 200
    data = response.json()
    assert len(data) == 2

四、项目功能总结

4.1 使用的 Claude Code 功能

功能 应用场景
Sub Agent 并行创建模型、路由、服务
MCP Browser 测试 API 端点
MCP Context7 查询 FastAPI/SQLAlchemy 文档
Hooks 自动化 运行测试、代码检查
Skills 标准化 API 代码生成
自然语言 Git 提交代码、创建分支

4.2 开发效率对比

任务 传统方式 Claude Code 效率提升
项目初始化 2 小时 20 分钟 6x
模型创建 3 小时 30 分钟 6x
API 开发 4 小时 1 小时 4x
测试编写 3 小时 30 分钟 6x
文档编写 2 小时 20 分钟 6x

五、下一步学习

完成本实战项目后,你已经掌握了 Claude Code 的核心功能。建议:

  1. 在自己的项目中应用这些技巧
  2. 创建更多自定义 Skills
  3. 探索更多 MCP 服务器
  4. 与 Cursor 结合使用,发挥各自优势

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐