api-design-expert

Expert guidance for API design, RESTful principles, GraphQL, versioning strategies, and API best practices.

Safety Notice

This listing is imported from skills.sh public index metadata. Review upstream SKILL.md and repository scripts before running.

Copy this and send it to your AI assistant to learn

Install skill "api-design-expert" with this command: npx skills add personamanagmentlayer/pcl/personamanagmentlayer-pcl-api-design-expert

API Design Expert

Expert guidance for API design, RESTful principles, GraphQL, versioning strategies, and API best practices.

Core Concepts

API Design Principles

  • RESTful architecture

  • Resource-oriented design

  • Uniform interface

  • Statelessness

  • Cacheability

  • Layered system

API Styles

  • REST (Representational State Transfer)

  • GraphQL

  • RPC (Remote Procedure Call)

  • WebSocket

  • Server-Sent Events (SSE)

  • gRPC

Key Considerations

  • Versioning strategies

  • Authentication and authorization

  • Rate limiting

  • Error handling

  • Documentation

  • Backward compatibility

REST API Design

from fastapi import FastAPI, HTTPException, Query, Path, Header from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime from enum import Enum

app = FastAPI( title="User Management API", version="1.0.0", description="RESTful API for user management" )

Models

class UserRole(str, Enum): ADMIN = "admin" USER = "user" GUEST = "guest"

class UserCreate(BaseModel): email: str = Field(..., example="user@example.com") name: str = Field(..., min_length=1, max_length=100) role: UserRole = UserRole.USER

class UserResponse(BaseModel): id: str email: str name: str role: UserRole created_at: datetime updated_at: datetime

class Config:
    schema_extra = {
        "example": {
            "id": "123e4567-e89b-12d3-a456-426614174000",
            "email": "user@example.com",
            "name": "John Doe",
            "role": "user",
            "created_at": "2024-01-01T00:00:00Z",
            "updated_at": "2024-01-01T00:00:00Z"
        }
    }

class UserUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=100) role: Optional[UserRole] = None

REST Endpoints following best practices

@app.get("/api/v1/users", response_model=List[UserResponse], summary="List all users", tags=["Users"]) async def list_users( page: int = Query(1, ge=1, description="Page number"), page_size: int = Query(20, ge=1, le=100, description="Items per page"), sort: str = Query("created_at", description="Sort field"), order: str = Query("desc", regex="^(asc|desc)$") ): """ Retrieve a paginated list of users.

- **page**: Page number (starts at 1)
- **page_size**: Number of items per page (1-100)
- **sort**: Field to sort by
- **order**: Sort order (asc or desc)
"""
# Implementation
return []

@app.get("/api/v1/users/{user_id}", response_model=UserResponse, summary="Get user by ID", tags=["Users"]) async def get_user( user_id: str = Path(..., description="User ID") ): """Retrieve a specific user by ID.""" # Implementation raise HTTPException(status_code=404, detail="User not found")

@app.post("/api/v1/users", response_model=UserResponse, status_code=201, summary="Create new user", tags=["Users"]) async def create_user(user: UserCreate): """Create a new user.""" # Implementation return UserResponse( id="123e4567-e89b-12d3-a456-426614174000", email=user.email, name=user.name, role=user.role, created_at=datetime.now(), updated_at=datetime.now() )

@app.patch("/api/v1/users/{user_id}", response_model=UserResponse, summary="Update user", tags=["Users"]) async def update_user( user_id: str = Path(..., description="User ID"), user: UserUpdate = None ): """Partially update a user.""" # Implementation pass

@app.delete("/api/v1/users/{user_id}", status_code=204, summary="Delete user", tags=["Users"]) async def delete_user( user_id: str = Path(..., description="User ID") ): """Delete a user.""" # Implementation pass

Nested resources

@app.get("/api/v1/users/{user_id}/posts", summary="Get user posts", tags=["Users", "Posts"]) async def get_user_posts(user_id: str): """Retrieve all posts for a specific user.""" return []

API Versioning

from fastapi import APIRouter, Request

URL Path Versioning (Recommended)

v1_router = APIRouter(prefix="/api/v1") v2_router = APIRouter(prefix="/api/v2")

@v1_router.get("/users") async def get_users_v1(): """Version 1: Returns basic user info""" return [{"id": 1, "name": "John"}]

@v2_router.get("/users") async def get_users_v2(): """Version 2: Returns enhanced user info""" return [{"id": 1, "name": "John", "email": "john@example.com"}]

Header Versioning

async def version_from_header(request: Request): version = request.headers.get("API-Version", "1") return version

@app.get("/api/users") async def get_users(version: str = Depends(version_from_header)): if version == "2": return get_users_v2() return get_users_v1()

Content Negotiation Versioning

@app.get("/api/users") async def get_users_content_negotiation( accept: str = Header("application/vnd.api+json; version=1") ): if "version=2" in accept: return get_users_v2() return get_users_v1()

Error Handling

from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel, ValidationError

class ErrorResponse(BaseModel): error: str message: str details: Optional[dict] = None timestamp: datetime path: str

@app.exception_handler(HTTPException) async def http_exception_handler(request: Request, exc: HTTPException): return JSONResponse( status_code=exc.status_code, content=ErrorResponse( error=exc.status_code, message=exc.detail, timestamp=datetime.now(), path=request.url.path ).dict() )

@app.exception_handler(ValidationError) async def validation_exception_handler(request: Request, exc: ValidationError): return JSONResponse( status_code=422, content=ErrorResponse( error="validation_error", message="Request validation failed", details=exc.errors(), timestamp=datetime.now(), path=request.url.path ).dict() )

Custom business logic errors

class BusinessError(Exception): def init(self, message: str, code: str): self.message = message self.code = code

@app.exception_handler(BusinessError) async def business_error_handler(request: Request, exc: BusinessError): return JSONResponse( status_code=400, content=ErrorResponse( error=exc.code, message=exc.message, timestamp=datetime.now(), path=request.url.path ).dict() )

Rate Limiting

from fastapi import Request, HTTPException from datetime import datetime, timedelta import redis from typing import Dict

class RateLimiter: def init(self, redis_client: redis.Redis): self.redis = redis_client

async def check_rate_limit(self,
                           key: str,
                           max_requests: int,
                           window_seconds: int) -> Dict:
    """
    Token bucket algorithm for rate limiting
    """
    now = datetime.now().timestamp()
    window_key = f"rate_limit:{key}:{int(now // window_seconds)}"

    pipe = self.redis.pipeline()
    pipe.incr(window_key)
    pipe.expire(window_key, window_seconds)
    result = pipe.execute()

    request_count = result[0]

    if request_count > max_requests:
        reset_time = (int(now // window_seconds) + 1) * window_seconds
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={
                "X-RateLimit-Limit": str(max_requests),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(reset_time)
            }
        )

    return {
        "limit": max_requests,
        "remaining": max_requests - request_count,
        "reset": (int(now // window_seconds) + 1) * window_seconds
    }

Middleware for rate limiting

@app.middleware("http") async def rate_limit_middleware(request: Request, call_next): limiter = RateLimiter(redis_client)

# Get client identifier (IP, API key, etc.)
client_id = request.client.host

try:
    rate_info = await limiter.check_rate_limit(
        client_id,
        max_requests=100,
        window_seconds=60
    )

    response = await call_next(request)

    # Add rate limit headers
    response.headers["X-RateLimit-Limit"] = str(rate_info["limit"])
    response.headers["X-RateLimit-Remaining"] = str(rate_info["remaining"])
    response.headers["X-RateLimit-Reset"] = str(rate_info["reset"])

    return response
except HTTPException as e:
    return JSONResponse(
        status_code=e.status_code,
        content={"error": e.detail},
        headers=e.headers
    )

HATEOAS Implementation

from typing import Dict, List

class HATEOASResponse(BaseModel): data: dict links: Dict[str, str]

def create_links(resource_id: str, resource_type: str) -> Dict[str, str]: """Create HATEOAS links for a resource""" return { "self": f"/api/v1/{resource_type}/{resource_id}", "update": f"/api/v1/{resource_type}/{resource_id}", "delete": f"/api/v1/{resource_type}/{resource_id}", "collection": f"/api/v1/{resource_type}" }

@app.get("/api/v1/users/{user_id}", response_model=HATEOASResponse) async def get_user_with_links(user_id: str): user = get_user_from_db(user_id)

return HATEOASResponse(
    data=user,
    links={
        "self": f"/api/v1/users/{user_id}",
        "posts": f"/api/v1/users/{user_id}/posts",
        "profile": f"/api/v1/users/{user_id}/profile",
        "update": f"/api/v1/users/{user_id}",
        "delete": f"/api/v1/users/{user_id}"
    }
)

API Security

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from jose import JWTError, jwt from passlib.context import CryptContext

security = HTTPBearer() pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

SECRET_KEY = "your-secret-key" ALGORITHM = "HS256"

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): """Verify JWT token""" try: payload = jwt.decode( credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM] ) return payload except JWTError: raise HTTPException( status_code=401, detail="Invalid authentication credentials" )

@app.get("/api/v1/protected") async def protected_route(token_payload: dict = Depends(verify_token)): return {"message": "Access granted", "user": token_payload}

API Key Authentication

async def verify_api_key(api_key: str = Header(...)): """Verify API key""" if api_key not in valid_api_keys: raise HTTPException(status_code=403, detail="Invalid API key") return api_key

Best Practices

Design

  • Use nouns for resources, not verbs

  • Use plural names for collections

  • Use HTTP methods correctly (GET, POST, PUT, PATCH, DELETE)

  • Return appropriate status codes

  • Support filtering, sorting, and pagination

  • Use consistent naming conventions

  • Version APIs from the start

Documentation

  • Use OpenAPI/Swagger

  • Provide example requests and responses

  • Document error codes and messages

  • Include authentication requirements

  • Keep documentation up-to-date

  • Provide SDKs when possible

Performance

  • Implement caching (ETags, Cache-Control)

  • Support compression (gzip)

  • Paginate large result sets

  • Use async operations for long tasks

  • Implement rate limiting

  • Monitor API performance

Security

  • Use HTTPS always

  • Implement authentication and authorization

  • Validate all inputs

  • Rate limit to prevent abuse

  • Log security events

  • Use API keys or OAuth 2.0

  • Implement CORS properly

Anti-Patterns

❌ Using GET for state-changing operations ❌ Returning inconsistent response formats ❌ No versioning strategy ❌ Poor error messages ❌ No rate limiting ❌ Exposing internal implementation details ❌ Breaking changes without versioning

Resources

Source Transparency

This detail page is rendered from real SKILL.md content. Trust labels are metadata-based hints, not a safety guarantee.

Related Skills

Related by shared tags or category signals.

General

finance-expert

No summary provided by upstream source.

Repository SourceNeeds Review
General

trading-expert

No summary provided by upstream source.

Repository SourceNeeds Review
General

dart-expert

No summary provided by upstream source.

Repository SourceNeeds Review