🚀 Advanced API Building — Production-Ready Services

Enterprise APIs require more than basic endpoints. Master async frameworks, database integration, caching strategies, comprehensive documentation, and deployment patterns.

⚡ FastAPI - Modern Async Framework

from fastapi import FastAPI, Path, Query, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
import asyncio

app = FastAPI(title='Advanced API', version='1.0')

# Pydantic models with validation
class User(BaseModel):
    id: int
    name: str = Field(..., min_length=1, max_length=100)
    email: str
    age: Optional[int] = Field(None, ge=0, le=150)

    class Config:
        schema_extra = {
            'example': {
                'id': 1,
                'name': 'Alice',
                'email': 'alice@example.com',
                'age': 28
            }
        }

# Async endpoints
@app.get('/users/{user_id}', response_model=User)
async def get_user(
    user_id: int = Path(..., gt=0, description='User ID'),
    include_posts: bool = Query(False, description='Include user posts')
):
    """Get user by ID with optional posts"""

    # Simulate async database call
    await asyncio.sleep(0.1)

    user = {'id': user_id, 'name': 'Alice', 'email': 'alice@example.com', 'age': 28}

    if not user:
        raise HTTPException(status_code=404, detail='User not found')

    return user

@app.post('/users', response_model=User, status_code=201)
async def create_user(user: User):
    """Create new user"""

    # Simulate async database insert
    await asyncio.sleep(0.1)

    return user

@app.get('/users', response_model=List[User])
async def list_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    """List users with pagination"""

    # Simulate async database query
    await asyncio.sleep(0.1)

    users = [
        {'id': i, 'name': f'User {i}', 'email': f'user{i}@example.com'}
        for i in range(1, 11)
    ]

    return users[skip:skip+limit]

💾 Database Integration with SQLAlchemy

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

# Database setup
DATABASE_URL = 'postgresql://user:password@localhost/dbname'
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

# SQLAlchemy model
class UserModel(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)

Base.metadata.create_all(bind=engine)

# Pydantic schema
class UserSchema(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        from_attributes = True

app = FastAPI()

def get_db():
    """Database dependency"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get('/users/{user_id}', response_model=UserSchema)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    """Get user from database"""

    user = db.query(UserModel).filter(UserModel.id == user_id).first()

    if not user:
        raise HTTPException(status_code=404, detail='User not found')

    return user

@app.post('/users', response_model=UserSchema)
async def create_user(user: UserSchema, db: Session = Depends(get_db)):
    """Create user in database"""

    db_user = UserModel(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)

    return db_user

🔄 Caching with Redis

import redis
import json
from functools import wraps
from datetime import timedelta

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_endpoint(expire: int = 300):
    """Decorator for endpoint caching"""

    def decorator(func):
        async def wrapper(*args, **kwargs):
            # Generate cache key
            cache_key = f'{func.__name__}:{str(args)}:{str(kwargs)}'

            # Check cache
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)

            # Call endpoint
            result = await func(*args, **kwargs)

            # Cache result
            redis_client.setex(
                cache_key,
                timedelta(seconds=expire),
                json.dumps(result, default=str)
            )

            return result

        return wrapper
    return decorator

# Usage
@app.get('/expensive-operation')
@cache_endpoint(expire=600)
async def expensive_operation():
    """Cached for 10 minutes"""

    # Simulate expensive computation
    await asyncio.sleep(2)

    return {'result': 'expensive computation result'}

📚 API Documentation with OpenAPI

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI(
    title='My Advanced API',
    description='A fully documented production API',
    version='1.0.0',
    openapi_url='/api/openapi.json',
    docs_url='/api/docs'
)

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title='My Advanced API',
        version='1.0.0',
        description='Complete API documentation',
        routes=app.routes,
    )

    openapi_schema['info']['x-logo'] = {
        'url': 'https://example.com/logo.png'
    }

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

# Endpoints with detailed documentation
@app.get(
    '/users/{user_id}',
    summary='Get user details',
    description='Retrieve detailed information about a specific user',
    tags=['users'],
    responses={
        200: {'description': 'User found'},
        404: {'description': 'User not found'}
    }
)
async def get_user(user_id: int):
    """
    Get a user by their ID.

    - **user_id**: The numeric ID of the user
    """
    return {'id': user_id, 'name': 'Alice'}

🔒 Security Middleware

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZIPMiddleware

# Add security middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=['https://example.com'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*']
)

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=['example.com', 'www.example.com']
)

app.add_middleware(
    GZIPMiddleware,
    minimum_size=1000
)

# Custom middleware
from starlette.middleware.base import BaseHTTPMiddleware
from time import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time()
        response = await call_next(request)
        elapsed = time() - start

        response.headers['X-Response-Time'] = str(elapsed)
        return response

app.add_middleware(TimingMiddleware)

🔍 Request/Response Logging

import logging
from fastapi import Request
from datetime import datetime

logger = logging.getLogger(__name__)

@app.middleware('http')
async def log_requests(request: Request, call_next):
    """Log all requests and responses"""

    # Request info
    request_id = request.headers.get('X-Request-ID', 'unknown')
    method = request.method
    path = request.url.path

    # Call endpoint
    start = time.time()
    response = await call_next(request)
    elapsed = time.time() - start

    # Log structured
    log_data = {
        'request_id': request_id,
        'method': method,
        'path': path,
        'status': response.status_code,
        'duration_ms': elapsed * 1000,
        'timestamp': datetime.utcnow().isoformat()
    }

    logger.info(json.dumps(log_data))

    return response

🧪 Comprehensive Testing

from fastapi.testclient import TestClient

client = TestClient(app)

def test_get_user():
    """Test getting user"""
    response = client.get('/users/1')

    assert response.status_code == 200
    assert response.json()['id'] == 1

def test_create_user():
    """Test creating user"""
    new_user = {
        'name': 'Bob',
        'email': 'bob@example.com',
        'age': 30
    }

    response = client.post('/users', json=new_user)

    assert response.status_code == 201
    assert response.json()['name'] == 'Bob'

def test_list_users():
    """Test listing users"""
    response = client.get('/users?skip=0&limit=10')

    assert response.status_code == 200
    assert isinstance(response.json(), list)

def test_invalid_user_id():
    """Test error handling"""
    response = client.get('/users/invalid')

    assert response.status_code == 422

📦 Deployment Configuration

# docker-compose.yml
version: '3.8'

services:
  api:
    image: python:3.11
    command: uvicorn main:app --host 0.0.0.0 --port 8000
    ports:
      - '8000:8000'
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/dbname
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis

  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=dbname
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    ports:
      - '6379:6379'

volumes:
  postgres_data:

✅ Key Takeaways

🔗 What's Next?

Learn advanced web scraping techniques.

Next: Advanced Web Scraping →

Ready for advanced challenges? Try advanced challenges