fastapi-microservices-development

What this skill does

# FastAPI Microservices Development

A comprehensive skill for building production-ready microservices using FastAPI. This skill covers REST API design patterns, asynchronous operations, dependency injection, testing strategies, and deployment best practices for scalable Python applications.

## When to Use This Skill

Use this skill when:

- Building RESTful microservices with Python
- Developing high-performance async APIs
- Creating production-grade web services with comprehensive validation
- Implementing service-oriented architectures
- Building APIs requiring advanced dependency injection
- Developing services with complex authentication/authorization
- Creating scalable, maintainable backend services
- Building APIs with automatic OpenAPI documentation
- Implementing WebSocket services alongside REST APIs
- Deploying containerized Python services to production

## Core Concepts

### FastAPI Fundamentals

FastAPI is a modern, high-performance web framework for building APIs with Python 3.7+ based on standard Python type hints.

**Key Features:**
- **Fast**: Very high performance, on par with NodeJS and Go (powered by Starlette and Pydantic)
- **Fast to code**: Increase development speed by 200-300%
- **Fewer bugs**: Reduce human-induced errors by about 40%
- **Intuitive**: Great editor support with autocompletion everywhere
- **Easy**: Designed to be easy to learn and use
- **Short**: Minimize code duplication
- **Robust**: Production-ready code with automatic interactive documentation
- **Standards-based**: Based on OpenAPI and JSON Schema

### Async/Await Programming

FastAPI fully supports asynchronous request handling using Python's `async`/`await` syntax:

```python
from fastapi import FastAPI

app = FastAPI()

@app.get('/burgers')
async def read_burgers():
    burgers = await get_burgers(2)
    return burgers
```

**When to use `async def`:**
- Database queries with async drivers
- External API calls
- File I/O operations
- Long-running computations that can be awaited
- WebSocket connections
- Background task processing

**When to use regular `def`:**
- Simple CRUD operations
- Synchronous database libraries
- CPU-bound operations
- Quick data transformations

### Dependency Injection System

FastAPI's dependency injection is one of its most powerful features, enabling:

- Code reusability across endpoints
- Shared logic implementation
- Database connection management
- Authentication and authorization
- Request validation
- Background task scheduling

**Basic Dependency Pattern:**

```python
from typing import Annotated, Union
from fastapi import Depends, FastAPI

app = FastAPI()

# Dependency function
async def common_parameters(
    q: Union[str, None] = None,
    skip: int = 0,
    limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}

# Using dependency in multiple endpoints
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return {"params": commons, "items": ["item1", "item2"]}

@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return {"params": commons, "users": ["user1", "user2"]}
```

## Microservices Architecture Patterns

### Service Design Principles

**1. Single Responsibility**
- Each microservice handles one business capability
- Clear boundaries and minimal coupling
- Independent deployment and scaling

**2. API-First Design**
- Design APIs before implementation
- Use OpenAPI schemas for contracts
- Version APIs appropriately

**3. Database Per Service**
- Each service owns its data
- No direct database sharing
- Use APIs for cross-service data access

**4. Stateless Services**
- Services don't maintain client session state
- Enables horizontal scaling
- Use external storage for session data

### Service Communication Patterns

**Synchronous Communication (REST APIs):**
```python
import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    # Call another microservice
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(f"http://inventory-service/stock/{order_id}")
            inventory_data = response.json()
        except httpx.HTTPError:
            raise HTTPException(status_code=503, detail="Inventory service unavailable")

    return {"order_id": order_id, "inventory": inventory_data}
```

**Event-Driven Communication:**
- Use message brokers (RabbitMQ, Kafka, Redis)
- Publish/Subscribe patterns
- Asynchronous processing
- Loose coupling between services

### Service Discovery

**Options:**
- Environment variables for simple setups
- Consul, Eureka for dynamic discovery
- Kubernetes DNS for K8s deployments
- API Gateway for centralized routing

## REST API Design Patterns

### Resource Modeling

**RESTful Resource Design:**

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

# Resource Models
class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        from_attributes = True

# Collection Endpoints
@app.get("/items/", response_model=List[Item])
async def list_items(skip: int = 0, limit: int = 100):
    """List all items with pagination"""
    items = await get_items_from_db(skip=skip, limit=limit)
    return items

@app.post("/items/", response_model=Item, status_code=201)
async def create_item(item: ItemCreate):
    """Create a new item"""
    new_item = await save_item_to_db(item)
    return new_item

# Resource Endpoints
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    """Get a specific item by ID"""
    item = await get_item_from_db(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: ItemCreate):
    """Update an existing item"""
    updated_item = await update_item_in_db(item_id, item)
    if updated_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return updated_item

@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    """Delete an item"""
    success = await delete_item_from_db(item_id)
    if not success:
        raise HTTPException(status_code=404, detail="Item not found")
```

### API Versioning

**URL Path Versioning (Recommended):**

```python
from fastapi import FastAPI, APIRouter

app = FastAPI()

# V1 API Router
v1_router = APIRouter(prefix="/api/v1")

@v1_router.get("/users/")
async def list_users_v1():
    return {"version": "v1", "users": []}

# V2 API Router
v2_router = APIRouter(prefix="/api/v2")

@v2_router.get("/users/")
async def list_users_v2():
    return {"version": "v2", "users": [], "metadata": {}}

app.include_router(v1_router)
app.include_router(v2_router)
```

### Request/Response Validation

FastAPI uses Pydantic for automatic validation:

```python
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=0, le=150)

    @validator('username')
    def username_alphanumeric(cls, v):
        assert v.isalnum(), 'must be alphanumeric'
        return v

    @validator('password')
    def password_strength(cls, v):
        if not any(char.isdigit() for char in v):
            raise ValueError('must contain at least one digit')
        if not any(char.isupper() for char in v):
            raise ValueError('must contain at least one uppercase letter')
        retur