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 (< 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 Round 3: Auth and access control 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 Round 4: Pagination, filtering, caching 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 Phase 3: Plan (ExitPlanMode) Write a concrete implementation plan covering: Files to create/modify — exact paths based on project structure discovered in Phase 1 Pydantic schemas — Create , Update , Response , and List schemas with field types SQLAlchemy model — table name, columns, relationships, indexes CRUD/service layer — async functions for each operation Router — endpoint signatures, status codes, response models Dependencies — auth, pagination, filtering dependencies Tests — test cases for happy path, validation errors, auth failures, not found Present via ExitPlanMode for user approval. Phase 4: Execute After approval, implement following this order: Step 1: Pydantic schemas 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 Step 2: SQLAlchemy model 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 ( ) ) Step 3: CRUD/service layer 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 Step 4: Router with dependencies 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 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 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 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).

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