Backend Development Specialist
Quick Reference (30 seconds)
Backend Development Mastery - Comprehensive backend development patterns covering API design, database integration, microservices, and modern architecture patterns.
Core Capabilities:
- API Design: REST, GraphQL, gRPC with OpenAPI 3.1
- Database Integration: PostgreSQL, MongoDB, Redis, caching strategies
- Microservices: Service mesh, distributed patterns, event-driven architecture
- Security: Authentication, authorization, OWASP compliance
- Performance: Caching, optimization, monitoring, scaling
When to Use:
- Backend API development and architecture
- Database design and optimization
- Microservices implementation
- Performance optimization and scaling
- Security integration for backend systems
Implementation Guide
API Design Patterns
RESTful API Architecture:
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI(title="Modern API", version="2.0.0")
security = HTTPBearer()
class UserResponse(BaseModel):
id: int
email: str
name: str
@app.get("/users", response_model=List[UserResponse])
async def list_users(token: str = Depends(security)):
"""List users with authentication."""
return await user_service.get_all_users()
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
"""Create new user with validation."""
return await user_service.create(user)GraphQL Implementation:
import strawberry
from typing import List
@strawberry.type
class User:
id: int
email: str
name: str
@strawberry.type
class Query:
@strawberry.field
async def users(self) -> List[User]:
return await user_service.get_all_users()
schema = strawberry.Schema(query=Query)Database Integration Patterns
PostgreSQL with SQLAlchemy:
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String, unique=True)
name = Column(String)
# Connection pooling and optimization
engine = create_engine(
DATABASE_URL,
pool_size=20,
max_overflow=30,
pool_pre_ping=True
)MongoDB with Motor:
from motor.motor_asyncio import AsyncIOMotorClient
from pymongo import IndexModel
class UserService:
def __init__(self, client: AsyncIOMotorClient):
self.db = client.myapp
self.users = self.db.users
# Index optimization
self.users.create_indexes([
IndexModel("email", unique=True),
IndexModel("created_at")
])
async def create_user(self, user_data: dict) -> str:
result = await self.users.insert_one(user_data)
return str(result.inserted_id)Microservices Architecture
Service Discovery with Consul:
import consul
class ServiceRegistry:
def __init__(self, consul_host="localhost", consul_port=8500):
self.consul = consul.Consul(host=consul_host, port=consul_port)
def register_service(self, service_name: str, service_id: str, port: int):
self.consul.agent.service.register(
name=service_name,
service_id=service_id,
port=port,
check=consul.Check.http(f"http://localhost:{port}/health", interval="10s")
)
def discover_service(self, service_name: str) -> List[str]:
_, services = self.consul.health.service(service_name, passing=True)
return [f"{s['Service']['Address']}:{s['Service']['Port']}" for s in services]Event-Driven Architecture:
import asyncio
from aio_pika import connect_robust
class EventBus:
def __init__(self, amqp_url: str):
self.connection = None
self.channel = None
self.amqp_url = amqp_url
async def connect(self):
self.connection = await connect_robust(self.amqp_url)
self.channel = await self.connection.channel()
async def publish_event(self, event_type: str, data: dict):
await self.channel.default_exchange.publish(
aio_pika.Message(
json.dumps({"type": event_type, "data": data}).encode(),
content_type="application/json"
),
routing_key=event_type
)Advanced Patterns
Caching Strategies
Redis Integration:
import redis.asyncio as redis
from functools import wraps
import json
import hashlib
class CacheManager:
def __init__(self, redis_url: str):
self.redis = redis.from_url(redis_url)
def cache_result(self, ttl: int = 3600):
def decorator(func):
@wraps(func)
async def wrapper(*args, kwargs):
cache_key = self._generate_cache_key(func.__name__, args, kwargs)
# Try to get from cache
cached = await self.redis.get(cache_key)
if cached:
return json.loads(cached)
# Execute function and cache result
result = await func(*args, kwargs)
await self.redis.setex(
cache_key,
ttl,
json.dumps(result, default=str)
)
return result
return wrapper
return decoratorSecurity Implementation
JWT Authentication:
import jwt
from datetime import datetime, timedelta
from passlib.context import CryptContext
class SecurityManager:
def __init__(self, secret_key: str):
self.secret_key = secret_key
self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(self, password: str) -> str:
return self.pwd_context.hash(password)
def verify_password(self, plain_password: str, hashed_password: str) -> bool:
return self.pwd_context.verify(plain_password, hashed_password)
def create_access_token(self, data: dict, expires_delta: timedelta = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, self.secret_key, algorithm="HS256")Performance Optimization
Database Connection Pooling:
from sqlalchemy.pool import QueuePool
from sqlalchemy import event
def create_optimized_engine(database_url: str):
engine = create_engine(
database_url,
poolclass=QueuePool,
pool_size=20,
max_overflow=30,
pool_pre_ping=True,
pool_recycle=3600,
echo=False
)
@event.listens_for(engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
context._query_start_time = time.time()
@event.listens_for(engine, "after_cursor_execute")
def receive_after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
total = time.time() - context._query_start_time
if total > 0.1: # Log slow queries
logger.warning(f"Slow query: {total:.2f}s - {statement[:100]}")
return engineWorks Well With
- moai-domain-frontend - Full-stack development integration
- moai-domain-database - Advanced database patterns
- moai-integration-mcp - MCP server development for backend services
- moai-quality-security - Security validation and compliance
- moai-foundation-core - Core architectural principles
Technology Stack
Primary Technologies:
- Languages: Python 3.13+, Node.js 20+, Go 1.23
- Frameworks: FastAPI, Django, Express.js, Gin
- Databases: PostgreSQL 16+, MongoDB 7+, Redis 7+
- Message Queues: RabbitMQ, Apache Kafka, Redis Pub/Sub
- Containerization: Docker, Kubernetes
- Monitoring: Prometheus, Grafana, OpenTelemetry
Integration Patterns:
- RESTful APIs with OpenAPI 3.1
- GraphQL with Apollo Federation
- gRPC for high-performance services
- Event-driven architecture with CQRS
- API Gateway patterns
- Circuit breakers and resilience patterns
Status: Production Ready Last Updated: 2025-11-30 Maintained by: MoAI-ADK Backend Team