[ PROMPT_NODE_27887 ]
Fastapi Endpoint
[ SKILL_DOCUMENTATION ]
# FastAPI Endpoint Builder
## When to use
Use this skill when you need to:
- Add new API endpoints to an existing FastAPI project
- Build CRUD operations with proper validation and error handling
- Set up authenticated endpoints with dependency injection
- Create async database queries with SQLAlchemy 2.0
- Generate complete test coverage for API routes
## Phase 1: Explore (Plan Mode)
Enter plan mode. Before writing any code, explore the existing project to understand:
### Project structure
- Find the FastAPI app entry point (`main.py`, `app.py`, or `app/__init__.py`)
- Identify the router organization pattern (single file vs `routers/` directory)
- Check for existing `models/`, `schemas/`, `crud/`, or `services/` directories
- Look at `pyproject.toml` or `requirements.txt` for installed dependencies
### Existing patterns
- How are existing endpoints structured? (function-based vs class-based)
- What ORM is used? (SQLAlchemy 2.0 async, Tortoise, raw SQL, none)
- How is the database session managed? (`Depends(get_db)`, middleware, other)
- What auth pattern exists? (OAuth2PasswordBearer, API key header, custom)
- Are there existing Pydantic base models or shared schemas?
- What response format is standard? (direct model, wrapped `{"data": ..., "meta": ...}`)
### Test patterns
- Where do tests live? (`tests/`, `test_*.py`, `*_test.py`)
- What test client is used? (httpx AsyncClient, TestClient, pytest-asyncio)
- Are there test fixtures for database and auth?
## Phase 2: Interview (AskUserQuestion)
Use AskUserQuestion to clarify requirements. Ask in rounds — do NOT dump all questions at once.
### Round 1: Core endpoint
```
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
```
### Round 2: Data model (if new resource)
```
Question: "What fields does the resource have? (describe briefly)"
Header: "Fields"
Options:
- "Simple ( 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 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
```
### Step 4: Router with dependencies
```python
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")
```
### Step 5: Tests
```python
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
```
## Key patterns to follow
### Dependency injection for auth
```python
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
```
### Cursor-based pagination helper
```python
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())
```
### Error responses
Always use FastAPI's `HTTPException` with consistent detail messages. For validation errors, Pydantic v2 handles them automatically via `RequestValidationError` (422).
```python
# 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")
```
## Checklist before finishing
- [ ] All endpoints return proper status codes (201 for POST, 204 for DELETE)
- [ ] Pydantic schemas use `model_config = ConfigDict(from_attributes=True)` for ORM mode
- [ ] List endpoint has pagination with configurable limit
- [ ] Auth dependency is applied to all non-public endpoints
- [ ] Tests cover: happy path, not found, unauthorized, validation errors
- [ ] Router is registered in the main FastAPI app
- [ ] Database model has proper indexes on filtered/sorted columns
Source: claude-code-templates (MIT). See About Us for full credits.