Claude Code Python全栈实战：FastAPI项目从零到部署

npm install -g @anthropic-ai/claude-code

claude --version
# 预期输出：claude 1.x.x (或更高版本)

flowchart LR
    A[认证方式选择] --> B[OAuth浏览器登录]
    A --> C[API Key环境变量]
    A --> D[API Key直接输入]

    B --> B1[运行 claude 自动弹出]
    C --> C1[export ANTHROPIC_API_KEY=sk-ant-xxx]
    D --> D1[首次运行时粘贴]

    style A fill:#7c3aed,color:#fff
    style B fill:#10b981,color:#fff
    style C fill:#10b981,color:#fff
    style D fill:#10b981,color:#fff

claude
# 浏览器自动打开，登录 Anthropic 账户即可

export ANTHROPIC_API_KEY="sk-ant-api03-xxxxx"
claude

# 查看帮助信息
claude --help

# 查看配置位置
claude config list

mkdir taskflow-api && cd taskflow-api
git init

claude

/init

# TaskFlow API

## 项目概述
任务管理 RESTful API，基于 Python FastAPI + SQLite 开发。

## 技术栈
- Python 3.11+
- FastAPI + Uvicorn
- SQLAlchemy 2.0 (async)
- SQLite (开发) / PostgreSQL (生产)
- Pytest + httpx (测试)

## 代码规范
- 使用 type hints，所有函数必须有参数和返回类型注解
- 遵循 PEP 8，单行最大 88 字符（Black 格式化）
- API 路由按资源分文件：routers/tasks.py, routers/users.py
- Pydantic v2 模型放在 schemas/ 目录
- 数据库模型放在 models/ 目录

## 项目结构

## 常用命令
- 运行服务：`uvicorn app.main:app --reload`
- 运行测试：`pytest -v`
- 格式化：`black . && isort .`
- 类型检查：`mypy app/`

## 禁止事项
- 不要修改 .env 文件
- 不要运行 drop table 或删除数据库操作
- 测试使用 SQLite 内存数据库，不要依赖外部服务

__pycache__/
*.pyc
.env
*.db
.venv/
.pytest_cache/
.mypy_cache/
.claude/

flowchart TB
    subgraph 交互模式
        IM[交互模式] --> PM[Plan模式]
        IM --> NM[正常模式]
        IM --> AM[Agent模式]
    end

    PM --> PM1["输入前先规划<br>/plan 切换"]
    NM --> NM1["边聊边写代码<br>默认模式"]
    AM --> AM1["自动执行多步任务<br>/goal 切换"]

    subgraph 核心操作
        S1["Shift+Tab 切换模式"]
        S2["Tab 自动补全"]
        S3["Esc 取消当前操作"]
        S4["Ctrl+C 中断生成"]
    end

    style IM fill:#7c3aed,color:#fff
    style PM fill:#3b82f6,color:#fff
    style NM fill:#10b981,color:#fff
    style AM fill:#f59e0b,color:#fff

请按照 CLAUDE.md 中定义的项目结构，初始化这个 FastAPI 项目。
包括：pyproject.toml、app/main.py、app/database.py、空的 __init__.py 文件。
使用 SQLAlchemy 2.0 async 模式 + aiosqlite。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(
    title="TaskFlow API",
    version="0.1.0",
    description="任务管理 RESTful API",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/health")
async def health_check() -> dict[str, str]:
    return {"status": "ok"}

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "sqlite+aiosqlite:///./taskflow.db"

engine = create_async_engine(DATABASE_URL, echo=False)
async_session = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def get_db() -> AsyncSession:
    async with async_session() as session:
        yield session

安装项目依赖并创建虚拟环境

python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn sqlalchemy aiosqlite pydantic

用 Poetry 初始化项目，添加 fastapi, uvicorn, sqlalchemy, aiosqlite 依赖

创建 Task 数据模型和对应的 Pydantic schema。
Task 需要字段：id, title, description, status (pending/in_progress/done), 
priority (low/medium/high), created_at, updated_at。
使用 SQLAlchemy 2.0 声明式模型和 Pydantic v2。

import enum
from datetime import datetime

from sqlalchemy import DateTime, Enum, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class TaskStatus(str, enum.Enum):
    pending = "pending"
    in_progress = "in_progress"
    done = "done"


class TaskPriority(str, enum.Enum):
    low = "low"
    medium = "medium"
    high = "high"


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    description: Mapped[str | None] = mapped_column(Text, default=None)
    status: Mapped[TaskStatus] = mapped_column(
        Enum(TaskStatus), default=TaskStatus.pending
    )
    priority: Mapped[TaskPriority] = mapped_column(
        Enum(TaskPriority), default=TaskPriority.medium
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

from datetime import datetime

from pydantic import BaseModel, Field

from app.models.task import TaskPriority, TaskStatus


class TaskCreate(BaseModel):
    title: str = Field(..., max_length=200)
    description: str | None = None
    status: TaskStatus = TaskStatus.pending
    priority: TaskPriority = TaskPriority.medium


class TaskUpdate(BaseModel):
    title: str | None = Field(None, max_length=200)
    description: str | None = None
    status: TaskStatus | None = None
    priority: TaskPriority | None = None


class TaskResponse(BaseModel):
    id: int
    title: str
    description: str | None
    status: TaskStatus
    priority: TaskPriority
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}

在 app/crud/task.py 中实现 Task 的完整 CRUD 操作。
使用 async/await，所有数据库操作都要用 async session。
包含：create_task, get_task, get_tasks (支持分页和状态过滤), 
update_task, delete_task。

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.task import Task, TaskStatus
from app.schemas.task import TaskCreate, TaskUpdate


async def create_task(db: AsyncSession, task_data: TaskCreate) -> Task:
    task = Task(**task_data.model_dump())
    db.add(task)
    await db.commit()
    await db.refresh(task)
    return task


async def get_task(db: AsyncSession, task_id: int) -> Task | None:
    return await db.get(Task, task_id)


async def get_tasks(
    db: AsyncSession,
    status: TaskStatus | None = None,
    skip: int = 0,
    limit: int = 20,
) -> tuple[list[Task], int]:
    query = select(Task)
    count_query = select(func.count()).select_from(Task)

    if status:
        query = query.where(Task.status == status)
        count_query = count_query.where(Task.status == status)

    total = (await db.execute(count_query)).scalar_one()
    result = await db.execute(
        query.order_by(Task.created_at.desc()).offset(skip).limit(limit)
    )
    return list(result.scalars().all()), total


async def update_task(
    db: AsyncSession, task_id: int, task_data: TaskUpdate
) -> Task | None:
    task = await get_task(db, task_id)
    if not task:
        return None
    for key, value in task_data.model_dump(exclude_unset=True).items():
        setattr(task, key, value)
    await db.commit()
    await db.refresh(task)
    return task


async def delete_task(db: AsyncSession, task_id: int) -> bool:
    task = await get_task(db, task_id)
    if not task:
        return False
    await db.delete(task)
    await db.commit()
    return True

创建 app/routers/tasks.py，实现完整的 RESTful API 路由。
使用 APIRouter，包含以下端点：
- POST /tasks — 创建任务
- GET /tasks — 列表（支持 ?status=pending&skip=0&limit=20）
- GET /tasks/{id} — 获取单个
- PUT /tasks/{id} — 更新
- DELETE /tasks/{id} — 删除

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

from app.crud import task as crud
from app.database import get_db
from app.models.task import TaskStatus
from app.schemas.task import TaskCreate, TaskResponse, TaskUpdate

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


@router.post("/", response_model=TaskResponse, status_code=201)
async def create_task(
    task_data: TaskCreate, db: AsyncSession = Depends(get_db)
) -> TaskResponse:
    task = await crud.create_task(db, task_data)
    return task


@router.get("/", response_model=list[TaskResponse])
async def list_tasks(
    status: TaskStatus | None = None,
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
) -> list[TaskResponse]:
    tasks, _ = await crud.get_tasks(db, status=status, skip=skip, limit=limit)
    return tasks


@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(task_id: int, db: AsyncSession = Depends(get_db)) -> TaskResponse:
    task = await crud.get_task(db, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@router.put("/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: int,
    task_data: TaskUpdate,
    db: AsyncSession = Depends(get_db),
) -> TaskResponse:
    task = await crud.update_task(db, task_id, task_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: int, db: AsyncSession = Depends(get_db)
) -> None:
    deleted = await crud.delete_task(db, task_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Task not found")

from app.routers import tasks

app.include_router(tasks.router)

from contextlib import asynccontextmanager

from app.database import Base, engine


@asynccontextmanager
async def lifespan(app):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(
    title="TaskFlow API",
    version="0.1.0",
    lifespan=lifespan,
)

启动 FastAPI 服务，确认能正常运行

uvicorn app.main:app --reload

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process
INFO:     Started server process
INFO:     Waiting for application startup.
INFO:     Application startup complete.

# 创建任务
curl -X POST http://127.0.0.1:8000/tasks/ \
  -H "Content-Type: application/json" \
  -d '{"title": "学习 Claude Code", "priority": "high"}'

# 获取任务列表
curl http://127.0.0.1:8000/tasks/

# 健康检查
curl http://127.0.0.1:8000/health

在 tests/ 目录下创建完整的测试文件 test_tasks.py。
使用 pytest + httpx 的 AsyncClient，SQLite 内存数据库。
覆盖所有 CRUD 操作的 happy path 和 error case。

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.database import Base, get_db
from app.main import app

TEST_DB_URL = "sqlite+aiosqlite://"

test_engine = create_async_engine(TEST_DB_URL, echo=False)
TestSession = async_sessionmaker(test_engine, expire_on_commit=False)


async def override_get_db() -> AsyncSession:
    async with TestSession() as session:
        yield session


app.dependency_overrides[get_db] = override_get_db


@pytest_asyncio.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
def base_url() -> str:
    return "http://test"


@pytest.mark.asyncio
async def test_create_task(base_url: str) -> None:
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url=base_url
    ) as client:
        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["status"] == "pending"


@pytest.mark.asyncio
async def test_list_tasks(base_url: str) -> None:
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url=base_url
    ) as client:
        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_not_found(base_url: str) -> None:
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url=base_url
    ) as client:
        response = await client.get("/tasks/999")
    assert response.status_code == 404


@pytest.mark.asyncio
async def test_update_task(base_url: str) -> None:
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url=base_url
    ) as client:
        create_resp = await client.post(
            "/tasks/", json={"title": "Original"}
        )
        task_id = create_resp.json()["id"]
        response = await client.put(
            f"/tasks/{task_id}",
            json={"title": "Updated", "status": "in_progress"},
        )
    assert response.status_code == 200
    assert response.json()["title"] == "Updated"


@pytest.mark.asyncio
async def test_delete_task(base_url: str) -> None:
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url=base_url
    ) as client:
        create_resp = await client.post(
            "/tasks/", json={"title": "To delete"}
        )
        task_id = create_resp.json()["id"]
        response = await client.delete(f"/tasks/{task_id}")
    assert response.status_code == 204

pip install pytest pytest-asyncio httpx
pytest -v

tests/test_tasks.py::test_create_task PASSED
tests/test_tasks.py::test_list_tasks PASSED
tests/test_tasks.py::test_get_task_not_found PASSED
tests/test_tasks.py::test_update_task PASSED
tests/test_tasks.py::test_delete_task PASSED

flowchart TB
    subgraph 权限层级
        L1["全局设置<br>~/.claude/settings.json"]
        L2["项目设置<br>.claude/settings.json"]
        L3["会话设置<br>/permissions 命令"]
    end

    L1 --> |"继承"| L2
    L2 --> |"覆盖"| L3

    subgraph 规则类型
        R1["allow — 自动允许"]
        R2["deny — 永远禁止"]
        R3["ask — 每次询问"]
    end

    L3 --> R1
    L3 --> R2
    L3 --> R3

    style L1 fill:#7c3aed,color:#fff
    style L2 fill:#3b82f6,color:#fff
    style L3 fill:#10b981,color:#fff

{
  "permissions": {
    "allow": [
      "Bash(python *)",
      "Bash(pytest *)",
      "Bash(uvicorn *)",
      "Bash(pip install *)",
      "Bash(black *)",
      "Bash(mypy *)",
      "Read",
      "Write",
      "Edit"
    ],
    "deny": [
      "Bash(rm -rf *)",
      "Bash(drop *)",
      "Bash(DROP *)"
    ]
  }
}

flowchart LR
    subgraph 上下文生命周期
        S[新会话] --> G[增长阶段]
        G --> C[compact压缩]
        C --> G2[继续增长]
        G2 --> C2[再次压缩]
        C2 --> E[会话结束]
    end

    subgraph 管理命令
        CMD1["/compact — 压缩历史"]
        CMD2["/clear — 清空会话"]
        CMD3["/cost — 查看费用"]
    end

    G -.-> CMD1
    C2 -.-> CMD2

    style S fill:#10b981,color:#fff
    style C fill:#f59e0b,color:#fff
    style E fill:#ef4444,color:#fff

taskflow-api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── task.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── task.py
│   ├── routers/
│   │   ├── __init__.py
│   │   └── tasks.py
│   └── crud/
│       ├── __init__.py
│       └── task.py
├── tests/
│   ├── __init__.py
│   └── test_tasks.py
├── .claude/
│   └── settings.json
├── .gitignore
├── CLAUDE.md
├── pyproject.toml
└── requirements.txt

给 Task 模型添加 User 关联。一个 User 可以有多个 Task。
创建完整的 User CRUD 和路由，包含用户注册（密码用 bcrypt 哈希）。
在 Task 路由中添加按用户过滤的功能。
更新测试覆盖新功能。

sequenceDiagram
    participant D as 开发者
    participant CC as Claude Code
    participant FS as 文件系统
    participant T as 终端

    D->>CC: "创建 FastAPI 项目骨架"
    CC->>FS: 创建目录和文件
    CC->>D: 展示创建的文件列表

    D->>CC: "定义 Task 数据模型"
    CC->>FS: 写入 models/task.py
    CC->>FS: 写入 schemas/task.py

    D->>CC: "实现 CRUD 和路由"
    CC->>FS: 写入 crud/task.py
    CC->>FS: 写入 routers/tasks.py

    D->>CC: "运行服务验证"
    CC->>T: uvicorn app.main:app --reload
    T-->>CC: 启动成功
    CC-->>D: 服务运行在 :8000

    D->>CC: "写测试并运行"
    CC->>FS: 写入 tests/test_tasks.py
    CC->>T: pytest -v
    T-->>CC: 5 passed
    CC-->>D: 所有测试通过 ✓

    D->>CC: /compact
    CC-->>D: 上下文已压缩

    D->>CC: "添加 User 模型和认证"
    Note over CC: 在压缩后的上下文中继续开发

# 按 Shift+Tab 切换到 Plan 模式
# 或者输入：
/plan

# 然后描述需求：
我需要给 API 添加 JWT 认证。请先规划实现方案，不要写代码。

## JWT 认证实现方案

1. 添加依赖：python-jose[cryptography], passlib[bcrypt]
2. 创建 app/core/security.py — JWT 编解码、密码哈希
3. 创建 app/core/config.py — 配置管理（SECRET_KEY 等）
4. 创建 app/routers/auth.py — 登录端点
5. 创建 app/deps.py — get_current_user 依赖
6. 在 Task 路由中添加认证保护

预估改动文件：7 个
预估时间：10 分钟

# 按 Shift+Tab 切换回正常模式
# 然后说：
按方案执行

请重新阅读 CLAUDE.md，然后按照其中的代码规范重写 app/routers/tasks.py

/compact 保留数据库模型定义和 API 路由结构，压缩其他历史

{
  "permissions": {
    "allow": [
      "Bash(python *)",
      "Bash(pip *)",
      "Bash(pytest *)",
      "Bash(uvicorn *)",
      "Bash(git *)",
      "Bash(ls *)",
      "Bash(cat *)",
      "Read",
      "Write",
      "Edit"
    ]
  }
}

- 测试必须使用 SQLite 内存数据库，不得依赖外部服务

创建一个多阶段 Dockerfile，基于 python:3.11-slim。
第一阶段安装依赖，第二阶段复制代码并运行 uvicorn。
暴露 8000 端口。

FROM python:3.11-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY . .

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

创建 docker-compose.yml，包含 api 服务和 PostgreSQL 数据库服务。
添加环境变量配置。

请帮我提交代码。提交信息要描述清楚本次创建的内容。

feat: init TaskFlow API with FastAPI

- Project skeleton with async SQLAlchemy + SQLite
- Task model with status and priority enums
- Full CRUD operations and RESTful routes
- Pydantic v2 schemas for request/response
- Async test suite with in-memory SQLite
- CLAUDE.md project configuration

Co-Authored-By: Claude Opus 4.6 <[email protected]>

flowchart TB
    START[开始新项目] --> INSTALL[安装 Claude Code<br>npm i -g @anthropic-ai/claude-code]
    INSTALL --> AUTH[认证<br>OAuth / API Key]
    AUTH --> INIT[初始化项目<br>/init → CLAUDE.md]
    INIT --> CONFIG[配置权限<br>.claude/settings.json]
    CONFIG --> DEV[开发循环]

    subgraph DEV[开发循环]
        direction TB
        PLAN[Plan模式<br>规划功能] --> CODE[正常模式<br>编写代码]
        CODE --> TEST[运行测试<br>pytest -v]
        TEST --> |失败| FIX[修复问题]
        FIX --> TEST
        TEST --> |通过| COMPACT[/compact<br>压缩上下文]
        COMPACT --> |下一个功能| PLAN
    end

    DEV --> DEPLOY[部署准备<br>Dockerfile + CI]

    style START fill:#7c3aed,color:#fff
    style DEV fill:#3b82f6,color:#fff
    style DEPLOY fill:#10b981,color:#fff