fastapi-endpoint

FastAPI 端点构建器

适用场景

在以下情况下使用此技能：

• 为现有 FastAPI 项目添加新的 API 端点
• 构建具备适当验证和错误处理的 CRUD 操作
• 使用依赖注入设置经过身份验证的端点
• 使用 SQLAlchemy 2.0 创建异步数据库查询
• 为 API 路由生成完整的测试覆盖

阶段 1：探索（计划模式）

进入计划模式。在编写任何代码之前，先探索现有项目，了解：

项目结构

• 找到 FastAPI 应用的入口点（main.py、app.py 或 app/__init__.py）
• 确定路由器的组织模式（单个文件还是 routers/ 目录）
• 检查是否存在 models/、schemas/、crud/ 或 services/ 目录
• 查看 pyproject.toml 或 requirements.txt 中已安装的依赖项

现有模式

• 现有端点采用什么结构？（基于函数还是基于类）
• 使用什么 ORM？（SQLAlchemy 2.0 异步、Tortoise、原始 SQL、无）
• 如何管理数据库会话？（Depends(get_db)、中间件或其他方式）
• 现有的身份验证模式是什么？（OAuth2PasswordBearer、API 密钥请求头、自定义方式）
• 是否存在 Pydantic 基础模型或共享架构？
• 标准的响应格式是什么？（直接返回模型，或封装为 {"data": ..., "meta": ...}）

测试模式

• 测试位于哪里？（tests/、test_*.py、*_test.py）
• 使用什么测试客户端？（httpx AsyncClient、TestClient、pytest-asyncio）
• 是否有用于数据库和身份验证的测试固件？

阶段 2：访谈（AskUserQuestion）

使用 AskUserQuestion 澄清需求。分轮提问，不要一次性抛出所有问题。

第 1 轮：核心端点

Question: "What resource does this endpoint manage?"
Header: "Resource"
Options:
  - "New resource (I'll describe the fields)" — Creating a new data model from scratch
  - "Existing model (extend it)" — Adding endpoints for a model that already exists in the codebase
  - "Relationship endpoint (nested)" — e.g., /users/{id}/orders — endpoint on a related resource

Question: "Which HTTP methods do you need?"
Header: "Methods"
multiSelect: true
Options:
  - "Full CRUD (GET list, GET detail, POST, PUT/PATCH, DELETE)" — All standard operations
  - "Read-only (GET list + GET detail)" — No mutations
  - "Custom action (POST /resource/{id}/action)" — Business logic endpoint, not standard CRUD

第 2 轮：数据模型（如为新资源）

Question: "What fields does the resource have? (describe briefly)"
Header: "Fields"
Options:
  - "Simple (< 6 fields, basic types)" — Strings, ints, booleans, dates
  - "Medium (6-15 fields, some relations)" — Includes foreign keys or enums
  - "Complex (nested objects, polymorphic)" — JSON fields, discriminated unions, computed fields

第 3 轮：身份验证和访问控制

Question: "How should this endpoint be authenticated?"
Header: "Auth"
Options:
  - "JWT Bearer token (Recommended)" — OAuth2PasswordBearer with JWT decode
  - "API Key header" — X-API-Key header validation
  - "No auth (public)" — Open endpoint, no authentication required
  - "Use existing auth" — Reuse the auth dependency already in the project

Question: "Do you need role-based access control?"
Header: "RBAC"
Options:
  - "No — any authenticated user" — Single permission level
  - "Yes — role check (admin, user, etc.)" — Require specific roles per endpoint
  - "Yes — ownership check" — Users can only access their own resources

第 4 轮：分页、筛选、缓存

Question: "What pagination style for list endpoints?"
Header: "Pagination"
Options:
  - "Cursor-based (Recommended)" — Best for real-time data, no offset drift
  - "Offset/limit" — Simple, good for admin panels with page numbers
  - "No pagination" — Small datasets, return all results

Question: "Do you need response caching?"
Header: "Caching"
Options:
  - "No caching" — Fresh data on every request
  - "Cache-Control headers" — Client-side caching via HTTP headers
  - "Redis/in-memory cache" — Server-side caching with TTL

阶段 3：制定计划（ExitPlanMode）

编写一份具体的实现计划，涵盖：

1. 要创建/修改的文件——根据阶段 1 中发现的项目结构列出准确路径
2. Pydantic 架构——包含字段类型的 Create、Update、Response 和 List 架构
3. SQLAlchemy 模型——表名、列、关系、索引
4. CRUD/服务层——每项操作对应的异步函数
5. 路由器——端点签名、状态码、响应模型
6. 依赖项——身份验证、分页、筛选依赖项
7. 测试——涵盖正常流程、验证错误、身份验证失败和未找到的测试用例

通过 ExitPlanMode 提交计划，供用户审批。

阶段 4：执行

获批后，按以下顺序实现：

步骤 1：Pydantic 架构

from pydantic import BaseModel, ConfigDict
from datetime import datetime
from uuid import UUID

class ResourceBase(BaseModel):
    """Shared fields between create and response."""
    name: str
    # ... fields from interview

class ResourceCreate(ResourceBase):
    """Fields required to create the resource."""
    pass

class ResourceUpdate(BaseModel):
    """All fields optional for partial updates."""
    name: str | None = None

class ResourceResponse(ResourceBase):
    """Full resource with DB-generated fields."""
    model_config = ConfigDict(from_attributes=True)
    id: UUID
    created_at: datetime
    updated_at: datetime

class ResourceListResponse(BaseModel):
    """Paginated list response."""
    data: list[ResourceResponse]
    next_cursor: str | None = None
    has_more: bool

步骤 2：SQLAlchemy 模型

from sqlalchemy import Column, String, DateTime, func
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
import uuid
from app.database import Base

class Resource(Base):
    __tablename__ = "resources"

    id = Column(PG_UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String, nullable=False, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

步骤 3：CRUD/服务层

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from uuid import UUID

async def get_resource(db: AsyncSession, resource_id: UUID) -> Resource | None:
    result = await db.execute(select(Resource).where(Resource.id == resource_id))
    return result.scalar_one_or_none()

async def list_resources(
    db: AsyncSession,
    cursor: str | None = None,
    limit: int = 20,
) -> tuple[list[Resource], str | None]:
    query = select(Resource).order_by(Resource.created_at.desc()).limit(limit + 1)
    if cursor:
        query = query.where(Resource.created_at < decode_cursor(cursor))
    result = await db.execute(query)
    items = list(result.scalars().all())
    next_cursor = encode_cursor(items[-1].created_at) if len(items) > limit else None
    return items[:limit], next_cursor

async def create_resource(db: AsyncSession, data: ResourceCreate) -> Resource:
    resource = Resource(**data.model_dump())
    db.add(resource)
    await db.commit()
    await db.refresh(resource)
    return resource

async def update_resource(
    db: AsyncSession, resource_id: UUID, data: ResourceUpdate
) -> Resource | None:
    resource = await get_resource(db, resource_id)
    if not resource:
        return None
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(resource, field, value)
    await db.commit()
    await db.refresh(resource)
    return resource

async def delete_resource(db: AsyncSession, resource_id: UUID) -> bool:
    resource = await get_resource(db, resource_id)
    if not resource:
        return False
    await db.delete(resource)
    await db.commit()
    return True

步骤 4：包含依赖项的路由器

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

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

@router.get("", response_model=ResourceListResponse)
async def list_resources_endpoint(
    cursor: str | None = Query(None),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),  # if auth required
):
    items, next_cursor = await list_resources(db, cursor=cursor, limit=limit)
    return ResourceListResponse(
        data=items,
        next_cursor=next_cursor,
        has_more=next_cursor is not None,
    )

@router.get("/{resource_id}", response_model=ResourceResponse)
async def get_resource_endpoint(
    resource_id: UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    resource = await get_resource(db, resource_id)
    if not resource:
        raise HTTPException(status_code=404, detail="Resource not found")
    return resource

@router.post("", response_model=ResourceResponse, status_code=status.HTTP_201_CREATED)
async def create_resource_endpoint(
    data: ResourceCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    return await create_resource(db, data)

@router.patch("/{resource_id}", response_model=ResourceResponse)
async def update_resource_endpoint(
    resource_id: UUID,
    data: ResourceUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    resource = await update_resource(db, resource_id, data)
    if not resource:
        raise HTTPException(status_code=404, detail="Resource not found")
    return resource

@router.delete("/{resource_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_resource_endpoint(
    resource_id: UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    deleted = await delete_resource(db, resource_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Resource not found")

步骤 5：测试

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_resource(client: AsyncClient, auth_headers: dict):
    response = await client.post(
        "/resources",
        json={"name": "Test Resource"},
        headers=auth_headers,
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Test Resource"
    assert "id" in data

@pytest.mark.asyncio
async def test_get_resource_not_found(client: AsyncClient, auth_headers: dict):
    response = await client.get(
        "/resources/00000000-0000-0000-0000-000000000000",
        headers=auth_headers,
    )
    assert response.status_code == 404

@pytest.mark.asyncio
async def test_list_resources_pagination(client: AsyncClient, auth_headers: dict):
    # Create multiple resources first
    for i in range(5):
        await client.post(
            "/resources",
            json={"name": f"Resource {i}"},
            headers=auth_headers,
        )
    response = await client.get("/resources?limit=2", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["data"]) == 2
    assert data["has_more"] is True
    assert data["next_cursor"] is not None

@pytest.mark.asyncio
async def test_create_resource_unauthorized(client: AsyncClient):
    response = await client.post("/resources", json={"name": "Test"})
    assert response.status_code in (401, 403)

@pytest.mark.asyncio
async def test_update_resource_partial(client: AsyncClient, auth_headers: dict):
    # Create
    create_resp = await client.post(
        "/resources",
        json={"name": "Original"},
        headers=auth_headers,
    )
    resource_id = create_resp.json()["id"]
    # Partial update
    response = await client.patch(
        f"/resources/{resource_id}",
        json={"name": "Updated"},
        headers=auth_headers,
    )
    assert response.status_code == 200
    assert response.json()["name"] == "Updated"

@pytest.mark.asyncio
async def test_delete_resource(client: AsyncClient, auth_headers: dict):
    create_resp = await client.post(
        "/resources",
        json={"name": "To Delete"},
        headers=auth_headers,
    )
    resource_id = create_resp.json()["id"]
    response = await client.delete(
        f"/resources/{resource_id}", headers=auth_headers
    )
    assert response.status_code == 204
    # Verify deleted
    get_resp = await client.get(
        f"/resources/{resource_id}", headers=auth_headers
    )
    assert get_resp.status_code == 404

应遵循的关键模式

用于身份验证的依赖注入

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = decode_jwt(token)
    user = await db.get(User, payload["sub"])
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

def require_role(*roles: str):
    """Factory for role-based access control."""
    async def checker(current_user: User = Depends(get_current_user)):
        if current_user.role not in roles:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return current_user
    return checker

基于游标的分页辅助函数

import base64
from datetime import datetime

def encode_cursor(dt: datetime) -> str:
    return base64.urlsafe_b64encode(dt.isoformat().encode()).decode()

def decode_cursor(cursor: str) -> datetime:
    return datetime.fromisoformat(base64.urlsafe_b64decode(cursor).decode())

错误响应

始终使用 FastAPI 的 HTTPException，并确保详情消息一致。对于验证错误，Pydantic v2 会通过 RequestValidationError 自动处理（422）。

# 404 — not found
raise HTTPException(status_code=404, detail="Resource not found")

# 409 — conflict (duplicate)
raise HTTPException(status_code=409, detail="Resource with this name already exists")

# 403 — forbidden
raise HTTPException(status_code=403, detail="Not allowed to modify this resource")

完成前检查清单

• 所有端点都返回正确的状态码（POST 返回 201，DELETE 返回 204）
• Pydantic 架构使用 model_config = ConfigDict(from_attributes=True) 启用 ORM 模式
• 列表端点支持分页，且分页上限可配置
• 所有非公开端点都应用了身份验证依赖项
• 测试涵盖：正常流程、未找到、未授权、验证错误
• 路由器已注册到主 FastAPI 应用中
• 数据库模型已在可筛选/排序的列上正确建立索引