学习目标：掌握用Claude Code从空白目录开始，一步步搭建完整的Python FastAPI REST API项目——包括现代工具链配置（uv + Ruff + pytest）、CLAUDE.md项目指令编写、分层架构设计、数据库集成、自动化测试、Docker容器化部署。读完本文，你能独立交付一个生产级FastAPI后端。
适用人群：有基础Python经验、想用AI编程助手加速后端开发的工程师。
| 工具 | 最低版本 | 安装命令 |
|---|---|---|
| Claude Code | 最新 | npm install -g @anthropic-ai/claude-code |
| Python | 3.11+ | 系统包管理器或 uv python install 3.12 |
| uv | 0.5+ | curl -LsSf https://astral.sh/uv/install.sh \| sh |
| Docker | 20+ | 系统包管理器 |
| Git | 2.40+ | 系统包管理器 |
确认环境就绪：
claude --version
uv --version
python --version
docker --version
传统 venv + pip 方式已落后。2026年Python生态的标准栈是 uv + Ruff + pytest——启动快10倍、依赖锁定精确、格式检查一体。
mkdir task-api && cd task-api
uv init --name task-api --python 3.12
uv init 生成以下结构：
task-api/
├── pyproject.toml
├── hello.py
└── .python-version
uv add fastapi uvicorn[standard] sqlalchemy alembic pydantic[email]
uv add --dev pytest pytest-asyncio httpx ruff
查看 pyproject.toml，确认依赖已写入：
[project]
name = "task-api"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"fastapi",
"uvicorn[standard]",
"sqlalchemy",
"alembic",
"pydantic[email]",
]
[dependency-groups]
dev = [
"pytest",
"pytest-asyncio",
"httpx",
"ruff",
]
在 pyproject.toml 末尾追加：
[tool.ruff]
target-version = "py312"
line-length = 100
[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM"]
[tool.ruff.format]
quote-style = "double"
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
CLAUDE.md是Claude Code的核心配置文件——它告诉AI你的项目结构、技术栈、编码规范和常用命令。一个精心编写的CLAUDE.md能让Claude Code的输出质量提升50%以上。
在项目根目录创建 CLAUDE.md：
# Task API - FastAPI Backend
## Tech Stack
- Python 3.12 + FastAPI + Uvicorn
- SQLAlchemy 2.0 (async) + Alembic
- Pydantic v2 for validation
- SQLite (dev) / PostgreSQL (prod)
- uv for package management
## Commands
- Run dev server: `uv run uvicorn app.main:app --reload`
- Run tests: `uv run pytest`
- Lint: `uv run ruff check .`
- Format: `uv run ruff format .`
- DB migration: `uv run alembic revision --autogenerate -m "desc"`
- DB upgrade: `uv run alembic upgrade head`
## Architecture
- `app/main.py` - FastAPI app entry, middleware, routers
- `app/models/` - SQLAlchemy ORM models
- `app/schemas/` - Pydantic request/response schemas
- `app/routers/` - API route handlers
- `app/services/` - Business logic layer
- `app/core/` - Config, database session, dependencies
- `tests/` - pytest test files (mirror app structure)
## Rules
- Always use async functions for database operations
- Use dependency injection for DB sessions (get_db)
- All endpoints must have type-annotated Pydantic schemas
- Follow RESTful naming: plural nouns for resources
- Include proper HTTP status codes and error responses
- Write tests for every new endpoint
- Use `uv run` prefix for all Python commands
- Never modify alembic migration files after they are committed
在 ~/.claude/CLAUDE.md 中添加Python通用规则：
## Python Preferences
- Use type hints everywhere (Python 3.12+ syntax)
- Prefer `from __future__ import annotations` for forward refs
- Use `async/await` for I/O-bound operations
- Follow PEP 695 for type aliases (`type Vector = list[float]`)
- Use f-strings, never .format() or %
- Sorting imports: stdlib → third-party → local
claude
进入交互模式后，输入第一个提示：
Create the full FastAPI project structure following the architecture defined in CLAUDE.md.
Include:
1. app/main.py with FastAPI app, CORS middleware, health check
2. app/core/config.py with Pydantic Settings
3. app/core/database.py with async SQLAlchemy engine and session
4. app/models/base.py with declarative base
5. app/schemas/health.py with health check response model
6. Empty __init__.py files in all packages
Claude Code会读取CLAUDE.md，生成符合你规范的项目骨架。
app/core/config.py —— 用Pydantic Settings管理配置：
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
app_name: str = "Task API"
version: str = "0.1.0"
database_url: str = "sqlite+aiosqlite:///./task.db"
debug: bool = True
model_config = {"env_file": ".env", "env_prefix": "APP_"}
settings = Settings()
app/core/database.py —— 异步数据库引擎：
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.core.config import settings
engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
app/main.py —— 应用入口：
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: create tables
async with app.state.engine.begin() as conn:
from app.models.base import Base
await conn.run_sync(Base.metadata.create_all)
yield
app = FastAPI(title=settings.app_name, version=settings.version, lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/health")
async def health_check():
return {"status": "ok", "app": settings.app_name, "version": settings.version}
在Claude Code中继续：
Start the dev server and verify the health check endpoint works
Claude Code执行：
uv run uvicorn app.main:app --reload
另开终端验证：
curl http://localhost:8000/health
# {"status":"ok","app":"Task API","version":"0.1.1"}
访问 http://localhost:8000/docs 可以看到Swagger UI自动文档。
向Claude Code发送：
Create a Task model in app/models/task.py with fields:
- id: UUID primary key
- title: string, required, max 200 chars
- description: string, optional
- status: enum (pending, in_progress, completed), default pending
- priority: enum (low, medium, high), default medium
- created_at: datetime with server default
- updated_at: datetime with onupdate trigger
Include Pydantic schemas for create, update, and response in app/schemas/task.py.
Register the model in models/__init__.py.
Claude Code生成以下关键代码：
app/models/task.py：
import uuid
from datetime import datetime
from sqlalchemy import String, text
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base
class Task(Base):
__tablename__ = "tasks"
id: Mapped[str] = mapped_column(
String(36), primary_key=True, default=lambda: str(uuid.uuid4())
)
title: Mapped[str] = mapped_column(String(200), nullable=False)
description: Mapped[str | None] = mapped_column(String(2000), nullable=True)
status: Mapped[str] = mapped_column(String(20), default="pending", nullable=False)
priority: Mapped[str] = mapped_column(String(20), default="medium", nullable=False)
created_at: Mapped[datetime] = mapped_column(
server_default=text("CURRENT_TIMESTAMP")
)
updated_at: Mapped[datetime | None] = mapped_column(
onupdate=datetime.utcnow, nullable=True
)
app/schemas/task.py：
import uuid
from datetime import datetime
from pydantic import BaseModel, Field
class TaskCreate(BaseModel):
title: str = Field(..., max_length=200)
description: str | None = Field(None, max_length=2000)
priority: str = Field("medium", pattern="^(low|medium|high)$")
class TaskUpdate(BaseModel):
title: str | None = Field(None, max_length=200)
description: str | None = Field(None, max_length=2000)
status: str | None = Field(None, pattern="^(pending|in_progress|completed)$")
priority: str | None = Field(None, pattern="^(low|medium|high)$")
class TaskResponse(BaseModel):
id: str
title: str
description: str | None
status: str
priority: str
created_at: datetime
updated_at: datetime | None
model_config = {"from_attributes": True}
app/services/task_service.py：
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.task import Task
from app.schemas.task import TaskCreate, TaskUpdate
class TaskService:
def __init__(self, db: AsyncSession):
self.db = db
async def create(self, data: TaskCreate) -> Task:
task = Task(**data.model_dump())
self.db.add(task)
await self.db.flush()
await self.db.refresh(task)
return task
async def list(self, status: str | None = None, skip: int = 0, limit: int = 50) -> list[Task]:
query = select(Task).offset(skip).limit(limit)
if status:
query = query.where(Task.status == status)
result = await self.db.execute(query)
return list(result.scalars().all())
async def get(self, task_id: str) -> Task | None:
return await self.db.get(Task, task_id)
async def update(self, task_id: str, data: TaskUpdate) -> Task | None:
task = await self.get(task_id)
if not task:
return None
for key, value in data.model_dump(exclude_unset=True).items():
setattr(task, key, value)
await self.db.flush()
await self.db.refresh(task)
return task
async def delete(self, task_id: str) -> bool:
task = await self.get(task_id)
if not task:
return False
await self.db.delete(task)
return True
app/routers/tasks.py：
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.schemas.task import TaskCreate, TaskResponse, TaskUpdate
from app.services.task_service import TaskService
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post("/", response_model=TaskResponse, status_code=201)
async def create_task(data: TaskCreate, db: AsyncSession = Depends(get_db)):
service = TaskService(db)
return await service.create(data)
@router.get("/", response_model=list[TaskResponse])
async def list_tasks(
status: str | None = None,
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
db: AsyncSession = Depends(get_db),
):
service = TaskService(db)
return await service.list(status=status, skip=skip, limit=limit)
@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(task_id: str, db: AsyncSession = Depends(get_db)):
service = TaskService(db)
task = await service.get(task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@router.patch("/{task_id}", response_model=TaskResponse)
async def update_task(task_id: str, data: TaskUpdate, db: AsyncSession = Depends(get_db)):
service = TaskService(db)
task = await service.update(task_id, data)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@router.delete("/{task_id}", status_code=204)
async def delete_task(task_id: str, db: AsyncSession = Depends(get_db)):
service = TaskService(db)
if not await service.delete(task_id):
raise HTTPException(status_code=404, detail="Task not found")
在 app/main.py 中注册路由：
from app.routers import tasks
app.include_router(tasks.router)
graph TB
Client[HTTP Client] -->|Request| Uvicorn[Uvicorn ASGI Server]
Uvicorn --> FastAPI[FastAPI App]
FastAPI --> CORS[CORS Middleware]
CORS --> Router[API Router /tasks]
Router -->|Dependency Injection| GetDB[get_db Session]
Router -->|Call| Service[TaskService]
Service -->|CRUD Operations| SQLAlchemy[SQLAlchemy Async]
SQLAlchemy -->|Async Engine| SQLite[(SQLite Dev DB)]
FastAPI -->|Auto-generated| Swagger[Swagger UI /docs]
FastAPI -->|Auto-generated| ReDoc[ReDoc /redoc]
subgraph Project Structure
A[app/main.py] --> B[app/routers/]
B --> C[app/services/]
C --> D[app/models/]
C --> E[app/schemas/]
A --> F[app/core/]
end
sequenceDiagram
participant C as Client
participant R as Router
participant S as TaskService
participant DB as Database
C->>R: POST /tasks {title: "Learn FastAPI"}
R->>R: Validate with TaskCreate schema
R->>S: service.create(data)
S->>DB: INSERT INTO tasks
DB-->>S: Return new task
S-->>R: Task ORM object
R->>R: Serialize with TaskResponse schema
R-->>C: 201 Created {id, title, status...}
C->>R: GET /tasks?status=pending
R->>S: service.list(status="pending")
S->>DB: SELECT * FROM tasks WHERE...
DB-->>S: Result rows
S-->>R: List of Task objects
R-->>C: 200 OK [tasks...]
C->>R: PATCH /tasks/{id} {status: "completed"}
R->>S: service.update(id, data)
S->>DB: UPDATE tasks SET status=...
DB-->>S: Updated task
S-->>R: Task object
R-->>C: 200 OK {updated task}
向Claude Code发送：
Create a complete test suite in tests/ following this pattern:
1. tests/conftest.py with async test client fixture and in-memory SQLite
2. tests/test_health.py for health endpoint
3. tests/test_tasks.py for full CRUD operations
Use pytest-asyncio and httpx.AsyncClient for testing.
tests/conftest.py：
import asyncio
from collections.abc import AsyncGenerator
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.core.database import get_db
from app.main import app
from app.models.base import Base
TEST_DB_URL = "sqlite+aiosqlite://"
test_engine = create_async_engine(TEST_DB_URL, echo=False)
TestSession = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
async with TestSession() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
@pytest.fixture(scope="session")
def event_loop():
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(autouse=True)
async def setup_db():
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
tests/test_tasks.py：
import pytest
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_create_task(client: AsyncClient):
response = await client.post("/tasks/", json={"title": "Test task", "priority": "high"})
assert response.status_code == 201
data = response.json()
assert data["title"] == "Test task"
assert data["priority"] == "high"
assert data["status"] == "pending"
assert "id" in data
@pytest.mark.asyncio
async def test_list_tasks(client: AsyncClient):
# Create 2 tasks
await client.post("/tasks/", json={"title": "Task 1"})
await client.post("/tasks/", json={"title": "Task 2"})
response = await client.get("/tasks/")
assert response.status_code == 200
assert len(response.json()) >= 2
@pytest.mark.asyncio
async def test_get_task(client: AsyncClient):
create_resp = await client.post("/tasks/", json={"title": "Find me"})
task_id = create_resp.json()["id"]
response = await client.get(f"/tasks/{task_id}")
assert response.status_code == 200
assert response.json()["title"] == "Find me"
@pytest.mark.asyncio
async def test_update_task(client: AsyncClient):
create_resp = await client.post("/tasks/", json={"title": "Original"})
task_id = create_resp.json()["id"]
response = await client.patch(f"/tasks/{task_id}", json={"status": "completed"})
assert response.status_code == 200
assert response.json()["status"] == "completed"
@pytest.mark.asyncio
async def test_delete_task(client: AsyncClient):
create_resp = await client.post("/tasks/", json={"title": "Delete me"})
task_id = create_resp.json()["id"]
response = await client.delete(f"/tasks/{task_id}")
assert response.status_code == 204
# Verify deletion
get_resp = await client.get(f"/tasks/{task_id}")
assert get_resp.status_code == 404
@pytest.mark.asyncio
async def test_404_not_found(client: AsyncClient):
response = await client.get("/tasks/nonexistent-id")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_validation_error(client: AsyncClient):
response = await client.post("/tasks/", json={"title": "x" * 201})
assert response.status_code == 422
在Claude Code中：
Run the full test suite and report results
uv run pytest -v
预期输出：
tests/test_health.py::test_health_check PASSED
tests/test_tasks.py::test_create_task PASSED
tests/test_tasks.py::test_list_tasks PASSED
tests/test_tasks.py::test_get_task PASSED
tests/test_tasks.py::test_update_task PASSED
tests/test_tasks.py::test_delete_task PASSED
tests/test_tasks.py::test_404_not_found PASSED
tests/test_tasks.py::test_validation_error PASSED
============ 8 passed in 1.24s ============
FROM python:3.12-slim AS builder
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-editable
FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /app/.venv /app/.venv
COPY app/ app/
ENV PATH="/app/.venv/bin:$PATH"
ENV APP_DATABASE_URL="sqlite+aiosqlite:///./task.db"
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
docker build -t task-api .
docker run -d -p 8000:8000 --name task-api task-api
验证：
docker exec task-api curl -s http://localhost:8000/health
services:
api:
build: .
ports:
- "8000:8000"
environment:
APP_DATABASE_URL: "postgresql+asyncpg://taskuser:taskpass@db:5432/taskdb"
depends_on:
db:
condition: service_healthy
db:
image: postgres:16-alpine
environment:
POSTGRES_USER: taskuser
POSTGRES_PASSWORD: taskpass
POSTGRES_DB: taskdb
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U taskuser -d taskdb"]
interval: 5s
timeout: 5s
retries: 5
volumes:
pgdata:
添加asyncpg依赖：
uv add asyncpg
启动完整栈：
docker compose up -d
下面用一个端到端场景演示API功能。启动服务后执行：
# 创建高优先级任务
curl -X POST http://localhost:8000/tasks/ \
-H "Content-Type: application/json" \
-d '{"title": "学习Claude Code", "description": "完成FastAPI教程", "priority": "high"}'
# 响应:
# {
# "id": "a1b2c3d4-...",
# "title": "学习Claude Code",
# "description": "完成FastAPI教程",
# "status": "pending",
# "priority": "high",
# "created_at": "2026-05-29T10:00:00",
# "updated_at": null
# }
# 查询所有高优先级任务
curl http://localhost:8000/tasks/?limit=10
# 更新任务状态为进行中
curl -X PATCH http://localhost:8000/tasks/a1b2c3d4-... \
-H "Content-Type: application/json" \
-d '{"status": "in_progress"}'
# 完成任务
curl -X PATCH http://localhost:8000/tasks/a1b2c3d4-... \
-H "Content-Type: application/json" \
-d '{"status": "completed"}'
# 删除任务
curl -X DELETE http://localhost:8000/tasks/a1b2c3d4-...
与Claude Code协作开发后端时，按以下层次逐步推进：
graph LR
A[第一轮: 骨架] --> B[第二轮: 模型+Schema]
B --> C[第三轮: Service层]
C --> D[第四轮: Router层]
D --> E[第五轮: 测试]
E --> F[第六轮: Docker部署]
style A fill:#e1f5fe
style F fill:#c8e6c9
每一轮的提示模板：
| 轮次 | 提示模式 | 示例 |
|---|---|---|
| 骨架 | "Create the full project structure following CLAUDE.md..." | 定义目录和基础文件 |
| 模型 | "Create a [Name] model with fields: ..." | 数据库表设计 |
| 服务 | "Implement the service layer for [Model] with CRUD operations" | 业务逻辑 |
| 路由 | "Create REST endpoints for [Model] using the service layer" | API暴露 |
| 测试 | "Write comprehensive tests for [Model] CRUD including edge cases" | 质量保障 |
| 部署 | "Create Dockerfile and docker-compose.yml for production" | 容器化 |
问题1：Claude Code生成的代码不遵循CLAUDE.md规范
解决方案：在提示中显式引用规则：
Following the rules in CLAUDE.md, create the Task model.
Remember: use async functions, type hints, and Pydantic v2 syntax.
问题2：数据库迁移失败
# 初始化Alembic
uv run alembic init alembic
# 在alembic/env.py中配置async模式
# 生成迁移
uv run alembic revision --autogenerate -m "add tasks table"
uv run alembic upgrade head
问题3：测试中出现 RuntimeError: Event loop is closed
在 conftest.py 中使用 scope="session" 的事件循环 fixture，并确保所有测试函数标记为 @pytest.mark.asyncio。
问题4：Docker中asyncpg导入失败
Dockerfile需要同时安装asyncpg：
uv add asyncpg
# 确保uv.lock已更新后再docker build
task-api/
├── CLAUDE.md # Claude Code项目配置
├── pyproject.toml # 项目元数据+依赖+工具配置
├── uv.lock # 精确依赖锁定
├── Dockerfile # 多阶段构建
├── docker-compose.yml # 完整开发环境
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI应用入口
│ ├── core/
│ │ ├── __init__.py
│ │ ├── config.py # Pydantic Settings
│ │ └── database.py # 异步数据库引擎
│ ├── models/
│ │ ├── __init__.py
│ │ ├── base.py # SQLAlchemy Base
│ │ └── task.py # Task ORM模型
│ ├── schemas/
│ │ ├── __init__.py
│ │ └── task.py # Pydantic schemas
│ ├── services/
│ │ ├── __init__.py
│ │ └── task_service.py # 业务逻辑
│ └── routers/
│ ├── __init__.py
│ └── tasks.py # API路由
└── tests/
├── __init__.py
├── conftest.py # 测试fixtures
├── test_health.py # 健康检查测试
└── test_tasks.py # CRUD完整测试
| 方向 | 下一步 |
|---|---|
| 认证 | 集成OAuth2 + JWT，用Claude Code生成auth中间件 |
| 分页 | 添加cursor-based分页到列表端点 |
| 缓存 | Redis集成，缓存频繁查询的端点 |
| 监控 | 添加Prometheus metrics中间件 |
| CI/CD | GitHub Actions自动测试+Docker构建 |
| API版本 | /api/v1/tasks 路由前缀管理 |
[!tip] 51学通信知识星球 星主：爱卫生（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 星球定位：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 当前规模：5200 篇主题 / 990 个回答，198 元/年。 加入福利：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 AI 加持：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。