fastapi-microservices-development
Comprehensive guide for building production-ready microservices with FastAPI including REST API patterns, async operations, dependency injection, and deployment strategies
- Category: discovered
- Tags: github, community
- Version: 1.0.0
- Author: walid
- Downloads: 0
- Updated: 2025-11-22
Download the latest version (v1.0.0) from this page.
SKILL.md
---
name: fastapi-microservices-development
description: Comprehensive guide for building production-ready microservices with FastAPI including REST API patterns, async operations, dependency injection, and deployment strategies
tags: [fastapi, microservices, rest-api, async, python, production]
tier: tier-1
---
# 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')
return v
class UserResponse(BaseModel):
id: int
username: str
email: EmailStr
created_at: datetime
class Config:
from_attributes = True
@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
# Automatic validation of request body
new_user = await save_user(user)
return new_user
```
## Advanced Dependency Injection
### Dependencies with Yield
Dependencies can use `yield` for setup/teardown operations:
```python
from fastapi import FastAPI, Depends
app = FastAPI()
# Database dependency with cleanup
async def get_db():
db = await connect_to_database()
try:
yield db
finally:
await db.close()
@app.get("/items/")
async def read_items(db = Depends(get_db)):
items = await db.query("SELECT * FROM items")
return items
```
**Advanced Resource Management:**
```python
from fastapi import Depends, HTTPException
async def get_database():
with Session() as session:
try:
yield session
except HTTPException:
session.rollback()
raise
finally:
session.close()
@app.post("/users/")
async def create_user(user: UserCreate, db = Depends(get_database)):
try:
new_user = db.add(User(**user.dict()))
db.commit()
return new_user
except Exception as e:
# Session automatically rolled back by dependency
raise HTTPException(status_code=500, detail=str(e))
```
### Sub-Dependencies
Dependencies can depend on other dependencies:
```python
from typing import Optional
from fastapi import FastAPI, Depends, Cookie
app = FastAPI()
async def query_extractor(q: Optional[str] = None):
return q
async def query_or_cookie_extractor(
q: str = Depends(query_extractor),
last_query: Optional[str] = Cookie(None)
):
if not q:
return last_query
return q
@app.get('/items/')
async def read_items(query: str = Depends(query_or_cookie_extractor)):
return {'query': query}
```
### Class-Based Dependencies
Use classes for complex dependency logic:
```python
from typing import Optional
from fastapi import FastAPI, Depends
app = FastAPI()
class CommonQueryParams:
def __init__(
self,
q: Optional[str] = None,
skip: int = 0,
limit: int = 100,
):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
# Shortcut syntax
@app.get("/users/")
async def read_users(commons: CommonQueryParams = Depends()):
return commons
```
### Global Dependencies
Apply dependencies to all routes:
```python
from fastapi import FastAPI, Depends, Header, HTTPException
async def verify_token(x_token: str = Header(...)):
if x_token != "secret-token":
raise HTTPException(status_code=400, detail="Invalid X-Token header")
return x_token
async def verify_key(x_key: str = Header(...)):
if x_key != "secret-key":
raise HTTPException(status_code=400, detail="Invalid X-Key header")
return x_key
# Apply to entire application
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
# Apply to router
from fastapi import APIRouter
router = APIRouter(
prefix="/items",
dependencies=[Depends(verify_token)]
)
@router.get("/")
async def read_items():
return [{"item_id": "Foo"}]
app.include_router(router)
```
### Reusable Dependency Aliases
Create type aliases for common dependencies:
```python
from typing import Annotated
from fastapi import Depends
# Define reusable dependency types
async def get_current_user():
return {"username": "johndoe"}
CurrentUser = Annotated[dict, Depends(get_current_user)]
# Use across multiple endpoints
@app.get("/items/")
def read_items(user: CurrentUser):
return {"user": user, "items": []}
@app.post("/items/")
def create_item(user: CurrentUser, item: Item):
return {"user": user, "item": item}
@app.delete("/items/{item_id}")
def delete_item(user: CurrentUser, item_id: int):
return {"user": user, "deleted": item_id}
```
## Authentication & Authorization
### OAuth2 with Password Flow
```python
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional
import jwt
from datetime import datetime, timedelta
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
class Token(BaseModel):
access_token: str
token_type: str
class User(BaseModel):
username: str
email: Optional[str] = None
full_name: Optional[str] = None
disabled: Optional[bool] = None
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except jwt.PyJWTError:
raise credentials_exception
user = await get_user_from_db(username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = await authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
```
### OAuth2 with Scopes
```python
from fastapi.security import SecurityScopes
from pydantic import ValidationError
async def get_current_user(
security_scopes: SecurityScopes,
token: str = Depends(oauth2_scheme)
):
if security_scopes.scopes:
authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
else:
authenticate_value = "Bearer"
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": authenticate_value},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
token_scopes = payload.get("scopes", [])
except (jwt.PyJWTError, ValidationError):
raise credentials_exception
for scope in security_scopes.scopes:
if scope not in token_scopes:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
headers={"WWW-Authenticate": authenticate_value},
)
user = await get_user(username)
if user is None:
raise credentials_exception
return user
@app.get("/items/", dependencies=[Security(get_current_user, scopes=["items:read"])])
async def read_items():
return [{"item_id": "Foo"}]
@app.post("/items/", dependencies=[Security(get_current_user, scopes=["items:write"])])
async def create_item(item: Item):
return item
```
## Background Tasks
### Simple Background Tasks
```python
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log_file:
log_file.write(message)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"Notification sent to {email}\n")
return {"message": "Notification sent in the background"}
```
### Background Tasks with Dependencies
```python
from fastapi import BackgroundTasks, Depends
from typing import Annotated
def write_log(message: str):
with open("log.txt", mode="a") as log_file:
log_file.write(message)
async def get_query_and_log(
query: str | None = None,
background_tasks: BackgroundTasks = Depends()
):
if query:
background_tasks.add_task(write_log, f"query: {query}\n")
return query
@app.post("/send-notification/{email}")
async def send_notification(
email: str,
background_tasks: BackgroundTasks,
query: Annotated[str | None, Depends(get_query_and_log)],
):
background_tasks.add_task(write_log, f"email: {email}, query: {query}\n")
return {"message": "Notification sent"}
```
## WebSocket Support
### Basic WebSocket
```python
from fastapi import WebSocket, WebSocketDisconnect
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message received: {data}")
except WebSocketDisconnect:
print("Client disconnected")
```
### WebSocket with Dependencies
```python
from fastapi import WebSocket, Depends, Query, Cookie, WebSocketException, status
async def get_cookie_or_token(
websocket: WebSocket,
session: str | None = Cookie(None),
token: str | None = Query(None),
):
if session is None and token is None:
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
return session or token
@app.websocket("/ws/{item_id}")
async def websocket_endpoint(
websocket: WebSocket,
item_id: str,
q: int | None = None,
cookie_or_token: str = Depends(get_cookie_or_token),
):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(
f"Session: {cookie_or_token}, Item: {item_id}, Data: {data}"
)
except WebSocketDisconnect:
print(f"Client {item_id} disconnected")
```
### WebSocket Connection Manager
```python
from typing import List
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in