API Design Expert Skill
When to Use This Skill
Activate when:
- •Creating FastAPI endpoints
- •Designing REST API routes
- •Implementing authentication
- •Adding error handling
- •Setting up rate limiting
- •Writing API documentation
Core Patterns
1. FastAPI Project Structure
code
src/api/
├── main.py # App entry point
├── config.py # Configuration
├── dependencies.py # Shared dependencies
├── models/ # Pydantic models
│ ├── creator.py
│ ├── video.py
│ └── attribution.py
├── routers/ # Route handlers
│ ├── creators.py
│ ├── videos.py
│ ├── attribution.py
│ └── analytics.py
├── services/ # Business logic
│ ├── creator_service.py
│ └── attribution_service.py
└── utils/ # Helpers
├── bigquery.py
└── auth.py
2. Main Application Setup
python
# src/api/main.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import time
from .routers import creators, videos, attribution, analytics
from .config import settings
app = FastAPI(
title="SponsorGraph API",
description="Creator marketing attribution platform",
version="1.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc"
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Request timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
# Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={
"status": "error",
"error": {
"code": "INTERNAL_SERVER_ERROR",
"message": "An unexpected error occurred",
"details": str(exc) if settings.DEBUG else None
}
}
)
# Include routers
app.include_router(creators.router, prefix="/api/v1", tags=["creators"])
app.include_router(videos.router, prefix="/api/v1", tags=["videos"])
app.include_router(attribution.router, prefix="/api/v1", tags=["attribution"])
app.include_router(analytics.router, prefix="/api/v1", tags=["analytics"])
# Health check
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"version": "1.0.0",
"timestamp": time.time()
}
3. Pydantic Models (Request/Response)
python
# src/api/models/creator.py
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Literal
from datetime import datetime
class CreatorBase(BaseModel):
"""Base creator fields"""
creator_id: str = Field(..., description="Unique creator identifier")
username: str = Field(..., min_length=1, max_length=100)
platform: Literal["youtube", "instagram", "tiktok"]
class CreatorCreate(CreatorBase):
"""Create creator request"""
follower_count: int = Field(..., ge=0, description="Number of followers")
is_verified: bool = False
@validator('follower_count')
def validate_follower_count(cls, v):
if v < 0:
raise ValueError('Follower count must be non-negative')
return v
class CreatorResponse(CreatorBase):
"""Creator response"""
follower_count: int
is_verified: bool
avg_engagement_rate: Optional[float] = Field(None, ge=0, le=1)
created_at: datetime
updated_at: datetime
class Config:
orm_mode = True
schema_extra = {
"example": {
"creator_id": "UCnKJeK_r90jDdIuzHXC0Org",
"username": "BobDoesSports",
"platform": "youtube",
"follower_count": 2100000,
"is_verified": True,
"avg_engagement_rate": 0.045,
"created_at": "2024-01-15T10:30:00Z",
"updated_at": "2026-02-07T15:22:00Z"
}
}
class CreatorListResponse(BaseModel):
"""Paginated creator list"""
status: str = "success"
data: List[CreatorResponse]
meta: dict = Field(..., description="Pagination metadata")
class Config:
schema_extra = {
"example": {
"status": "success",
"data": [...], # List of creators
"meta": {
"total": 150,
"page": 1,
"per_page": 20,
"total_pages": 8
}
}
}
4. REST API Endpoints
python
# src/api/routers/creators.py
from fastapi import APIRouter, Depends, HTTPException, Query
from typing import Optional, List
from ..models.creator import CreatorResponse, CreatorListResponse, CreatorCreate
from ..services.creator_service import CreatorService
from ..dependencies import get_creator_service, get_current_user
router = APIRouter()
@router.get("/creators", response_model=CreatorListResponse)
async def list_creators(
platform: Optional[str] = Query(None, description="Filter by platform"),
min_followers: Optional[int] = Query(None, ge=0, description="Minimum followers"),
page: int = Query(1, ge=1, description="Page number"),
per_page: int = Query(20, ge=1, le=100, description="Items per page"),
service: CreatorService = Depends(get_creator_service)
):
"""
List creators with optional filtering and pagination
- **platform**: Filter by social media platform
- **min_followers**: Minimum follower count
- **page**: Page number (1-indexed)
- **per_page**: Results per page (max 100)
"""
creators, total = await service.list_creators(
platform=platform,
min_followers=min_followers,
page=page,
per_page=per_page
)
return {
"status": "success",
"data": creators,
"meta": {
"total": total,
"page": page,
"per_page": per_page,
"total_pages": (total + per_page - 1) // per_page
}
}
@router.get("/creators/{creator_id}", response_model=CreatorResponse)
async def get_creator(
creator_id: str,
service: CreatorService = Depends(get_creator_service)
):
"""
Get creator by ID
Returns detailed creator information including engagement metrics
"""
creator = await service.get_creator(creator_id)
if not creator:
raise HTTPException(
status_code=404,
detail={
"code": "CREATOR_NOT_FOUND",
"message": f"Creator with ID '{creator_id}' not found"
}
)
return creator
@router.post("/creators", response_model=CreatorResponse, status_code=201)
async def create_creator(
creator: CreatorCreate,
service: CreatorService = Depends(get_creator_service),
current_user = Depends(get_current_user) # Requires authentication
):
"""
Create a new creator
Requires authentication. Only admin users can create creators.
"""
if not current_user.is_admin:
raise HTTPException(
status_code=403,
detail={
"code": "FORBIDDEN",
"message": "Admin access required"
}
)
return await service.create_creator(creator)
@router.get("/creators/{creator_id}/stats")
async def get_creator_stats(
creator_id: str,
days: int = Query(30, ge=1, le=365, description="Days of data"),
service: CreatorService = Depends(get_creator_service)
):
"""
Get creator performance statistics
Returns aggregated metrics over the specified time period
"""
stats = await service.get_creator_stats(creator_id, days=days)
if not stats:
raise HTTPException(
status_code=404,
detail={
"code": "CREATOR_NOT_FOUND",
"message": f"Creator with ID '{creator_id}' not found"
}
)
return {
"status": "success",
"data": stats,
"meta": {
"creator_id": creator_id,
"days": days
}
}
5. Service Layer (Business Logic)
python
# src/api/services/creator_service.py
from google.cloud import bigquery
from typing import Optional, List, Tuple
from ..models.creator import CreatorResponse, CreatorCreate
class CreatorService:
"""Business logic for creator operations"""
def __init__(self, bq_client: bigquery.Client):
self.bq = bq_client
async def list_creators(
self,
platform: Optional[str] = None,
min_followers: Optional[int] = None,
page: int = 1,
per_page: int = 20
) -> Tuple[List[CreatorResponse], int]:
"""
List creators with filtering and pagination
Returns:
(creators, total_count)
"""
# Build query with filters
where_clauses = []
if platform:
where_clauses.append(f"platform = '{platform}'")
if min_followers:
where_clauses.append(f"follower_count >= {min_followers}")
where_sql = " AND ".join(where_clauses) if where_clauses else "1=1"
# Get total count
count_query = f"""
SELECT COUNT(*) as total
FROM `mart.dim_creator`
WHERE {where_sql}
"""
total = self.bq.query(count_query).to_dataframe().iloc[0]['total']
# Get paginated results
offset = (page - 1) * per_page
data_query = f"""
SELECT
creator_id,
username,
platform,
follower_count,
is_verified,
avg_engagement_rate,
created_at,
updated_at
FROM `mart.dim_creator`
WHERE {where_sql}
ORDER BY follower_count DESC
LIMIT {per_page}
OFFSET {offset}
"""
df = self.bq.query(data_query).to_dataframe()
creators = [CreatorResponse(**row) for row in df.to_dict('records')]
return creators, total
async def get_creator(self, creator_id: str) -> Optional[CreatorResponse]:
"""Get creator by ID"""
query = f"""
SELECT
creator_id,
username,
platform,
follower_count,
is_verified,
avg_engagement_rate,
created_at,
updated_at
FROM `mart.dim_creator`
WHERE creator_id = @creator_id
"""
job_config = bigquery.QueryJobConfig(
query_parameters=[
bigquery.ScalarQueryParameter("creator_id", "STRING", creator_id)
]
)
df = self.bq.query(query, job_config=job_config).to_dataframe()
if df.empty:
return None
return CreatorResponse(**df.iloc[0].to_dict())
async def get_creator_stats(
self,
creator_id: str,
days: int = 30
) -> Optional[dict]:
"""Get creator performance stats"""
query = f"""
SELECT
SUM(views) as total_views,
AVG(engagement_rate) as avg_engagement,
COUNT(DISTINCT video_id) as video_count,
SUM(revenue_usd) as total_revenue
FROM `mart.fact_video_performance`
WHERE creator_id = @creator_id
AND published_date >= DATE_SUB(CURRENT_DATE(), INTERVAL @days DAY)
"""
job_config = bigquery.QueryJobConfig(
query_parameters=[
bigquery.ScalarQueryParameter("creator_id", "STRING", creator_id),
bigquery.ScalarQueryParameter("days", "INT64", days)
]
)
df = self.bq.query(query, job_config=job_config).to_dataframe()
if df.empty:
return None
return df.iloc[0].to_dict()
6. Error Handling
python
# src/api/models/error.py
from pydantic import BaseModel
from typing import Optional
class ErrorResponse(BaseModel):
"""Standard error response"""
status: str = "error"
error: dict
class Config:
schema_extra = {
"example": {
"status": "error",
"error": {
"code": "CREATOR_NOT_FOUND",
"message": "Creator with ID 'abc123' not found",
"details": None
}
}
}
# Custom exceptions
class CreatorNotFoundError(Exception):
"""Raised when creator is not found"""
pass
class RateLimitError(Exception):
"""Raised when rate limit is exceeded"""
pass
# Exception handlers
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
@app.exception_handler(CreatorNotFoundError)
async def creator_not_found_handler(request: Request, exc: CreatorNotFoundError):
return JSONResponse(
status_code=404,
content={
"status": "error",
"error": {
"code": "CREATOR_NOT_FOUND",
"message": str(exc)
}
}
)
@app.exception_handler(RateLimitError)
async def rate_limit_handler(request: Request, exc: RateLimitError):
return JSONResponse(
status_code=429,
content={
"status": "error",
"error": {
"code": "RATE_LIMIT_EXCEEDED",
"message": "Too many requests. Please try again later.",
"retry_after": 60
}
}
)
7. Authentication
python
# src/api/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from .config import settings
security = HTTPBearer()
def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
"""
Validate JWT token and return current user
"""
token = credentials.credentials
try:
payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms=["HS256"]
)
user_id: str = payload.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={
"code": "INVALID_TOKEN",
"message": "Invalid authentication token"
}
)
return {
"user_id": user_id,
"email": payload.get("email"),
"is_admin": payload.get("is_admin", False)
}
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={
"code": "INVALID_TOKEN",
"message": "Invalid authentication token"
}
)
8. Rate Limiting
python
# src/api/utils/rate_limit.py
from fastapi import Request, HTTPException
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
# In main.py:
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# In route:
@router.get("/creators")
@limiter.limit("100/minute")
async def list_creators(request: Request, ...):
...
REST API Conventions
URL Patterns
python
# Good: RESTful, predictable
GET /api/v1/creators # List creators
GET /api/v1/creators/{id} # Get creator
POST /api/v1/creators # Create creator
PUT /api/v1/creators/{id} # Update creator
DELETE /api/v1/creators/{id} # Delete creator
GET /api/v1/creators/{id}/videos # Get creator's videos
# Bad: Non-RESTful
GET /api/v1/getCreators
POST /api/v1/creator/create
GET /api/v1/creator?action=list
HTTP Status Codes
python
# Success 200 OK # GET, PUT, DELETE success 201 Created # POST success 204 No Content # DELETE success (no body) # Client Errors 400 Bad Request # Invalid input 401 Unauthorized # Missing/invalid auth token 403 Forbidden # Valid token, insufficient permissions 404 Not Found # Resource doesn't exist 409 Conflict # Resource already exists 422 Unprocessable Entity # Validation error 429 Too Many Requests # Rate limit exceeded # Server Errors 500 Internal Server Error # Unexpected error 503 Service Unavailable # Temporary outage
Response Format
python
# Success response
{
"status": "success",
"data": {...},
"meta": {
"timestamp": "2026-02-07T15:30:00Z",
"version": "v1"
}
}
# Error response
{
"status": "error",
"error": {
"code": "CREATOR_NOT_FOUND",
"message": "Creator with ID 'abc123' not found",
"details": {...} # Optional
}
}
Testing
python
# tests/api/test_creators.py
from fastapi.testclient import TestClient
from src.api.main import app
client = TestClient(app)
def test_list_creators():
"""Test GET /creators"""
response = client.get("/api/v1/creators")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert "data" in data
assert "meta" in data
def test_get_creator_not_found():
"""Test GET /creators/{id} with invalid ID"""
response = client.get("/api/v1/creators/invalid_id")
assert response.status_code == 404
data = response.json()
assert data["status"] == "error"
assert data["error"]["code"] == "CREATOR_NOT_FOUND"
def test_create_creator_unauthorized():
"""Test POST /creators without auth"""
response = client.post(
"/api/v1/creators",
json={
"creator_id": "test123",
"username": "testuser",
"platform": "youtube",
"follower_count": 1000
}
)
assert response.status_code == 401
Related Resources
- •Main context:
.claude/CLAUDE.md - •FastAPI docs: https://fastapi.tiangolo.com
- •Pydantic docs: https://docs.pydantic.dev