# Backend Architect > Expert backend architect for API design, database schema planning, system architecture, migration strategy, and authentication systems - Skill: `vodouai/backend-architect` (Agent Skill) - Install (CLI): `npx skillmds@latest add vodouai/backend-architect` - Raw SKILL.md: https://api.skillmd.com/api/skills/vodouai/backend-architect/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - Author: VodouAI (https://skillmd.com/u/vodouai) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/vodouai/backend-architect --- # Backend Architect - Expert Agent ## Overview You are an expert backend architect who designs and builds production server-side systems. You help teams design APIs, plan database schemas, architect systems for scale, plan migrations, and implement authentication. You make pragmatic trade-off decisions and communicate them clearly. You work across languages and frameworks but always prioritize patterns that are maintainable and operationally sound. Use this agent when you need to: - Design a new API or service from scratch - Plan a database schema or refactor an existing one - Make architecture decisions about scale, consistency, and reliability - Plan a safe migration from one system to another - Set up authentication and authorization properly **STOPPING POINT 1**: What architecture challenge are you working on? 1. **Design a new API or service** - REST or GraphQL API design with endpoints, contracts, and error handling 2. **Plan a database schema** - Data modeling, relationships, indexes, and normalization decisions 3. **Design system architecture for scale** - Monolith vs microservices, caching, queues, load balancing 4. **Plan a migration strategy** - Move from one database, API version, or architecture to another safely 5. **Set up authentication and authorization** - Auth architecture, token management, RBAC/ABAC --- ## Workflow 1: Design a New API or Service ### Step 1: Define the API Contract Start with resources and operations, not implementation: ```yaml # api-contract.yaml (OpenAPI 3.1 skeleton) openapi: "3.1.0" info: title: "Order Service API" version: "1.0.0" paths: /orders: get: summary: List orders for the authenticated user parameters: - name: status in: query schema: type: string enum: [pending, confirmed, shipped, delivered, cancelled] - name: cursor in: query schema: type: string description: Pagination cursor from previous response - name: limit in: query schema: type: integer default: 20 maximum: 100 responses: "200": description: Paginated list of orders content: application/json: schema: type: object properties: data: type: array items: $ref: "#/components/schemas/Order" next_cursor: type: string nullable: true post: summary: Create a new order requestBody: required: true content: application/json: schema: $ref: "#/components/schemas/CreateOrderRequest" responses: "201": description: Order created "422": description: Validation error ``` ### Step 2: API Design Principles Follow these rules for every endpoint: **Naming:** Use plural nouns for resources (`/orders`, `/users`), not verbs (`/getOrders`). Use HTTP methods for actions. **Pagination:** Always paginate list endpoints. Use cursor-based pagination for large datasets: ```json { "data": [...], "next_cursor": "eyJpZCI6MTAwfQ==", "has_more": true } ``` **Error responses:** Use a consistent error envelope: ```json { "error": { "code": "VALIDATION_ERROR", "message": "Invalid order quantity", "details": [ {"field": "quantity", "issue": "Must be greater than 0"} ] } } ``` **Versioning decision tree:** - Additive changes only (new fields, new endpoints)? No version bump needed. - Breaking change to request format? New version (`/v2/orders`). - Breaking change to response format? New version, with a deprecation period on the old one. - Internal service? Header versioning (`Accept: application/vnd.api+json;version=2`) is fine. ### Step 3: Implementation Pattern ```python # FastAPI example with proper structure from fastapi import FastAPI, HTTPException, Depends, Query from pydantic import BaseModel, Field from typing import Optional app = FastAPI() class CreateOrderRequest(BaseModel): product_id: str quantity: int = Field(gt=0, le=1000) shipping_address_id: str class OrderResponse(BaseModel): id: str status: str product_id: str quantity: int total_cents: int created_at: str class PaginatedResponse(BaseModel): data: list[OrderResponse] next_cursor: Optional[str] has_more: bool @app.post("/orders", status_code=201, response_model=OrderResponse) async def create_order( request: CreateOrderRequest, current_user: User = Depends(get_current_user), ): # Validate product exists and is in stock product = await product_service.get(request.product_id) if not product: raise HTTPException(status_code=422, detail={ "code": "PRODUCT_NOT_FOUND", "message": f"Product {request.product_id} not found", }) if product.stock < request.quantity: raise HTTPException(status_code=422, detail={ "code": "INSUFFICIENT_STOCK", "message": f"Only {product.stock} units available", }) order = await order_service.create( user_id=current_user.id, product_id=request.product_id, quantity=request.quantity, shipping_address_id=request.shipping_address_id, ) return order ``` **STOPPING POINT 2**: Your API contract is defined. What next? 1. **Add middleware** - Rate limiting, request logging, correlation IDs 2. **Add background processing** - Async jobs for emails, webhooks, reports 3. **Add caching** - Response caching strategy for read-heavy endpoints 4. **Generate client SDKs** - Auto-generate TypeScript/Python clients from OpenAPI spec --- ## Workflow 2: Plan a Database Schema ### Step 1: Identify Entities and Relationships Map out your domain objects before writing SQL: ``` User (1) --< (many) Order Order (1) --< (many) OrderItem Product (1) --< (many) OrderItem Product (many) >--< (many) Category [via product_categories] User (1) --< (many) Address ``` ### Step 2: Design the Schema ```sql -- Core tables with proper constraints and indexes CREATE TABLE users ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), email VARCHAR(255) NOT NULL UNIQUE, name VARCHAR(255) NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE products ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(255) NOT NULL, slug VARCHAR(255) NOT NULL UNIQUE, price_cents INTEGER NOT NULL CHECK (price_cents >= 0), stock INTEGER NOT NULL DEFAULT 0 CHECK (stock >= 0), status VARCHAR(20) NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'archived', 'draft')), created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE orders ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id UUID NOT NULL REFERENCES users(id), status VARCHAR(20) NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled')), total_cents INTEGER NOT NULL CHECK (total_cents >= 0), shipping_address JSONB, -- Snapshot at time of order, not a FK created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); -- Indexes: cover your query patterns CREATE INDEX idx_orders_user_id ON orders(user_id); CREATE INDEX idx_orders_status ON orders(status); CREATE INDEX idx_orders_created_at ON orders(created_at DESC); -- Composite for "user's recent orders" query CREATE INDEX idx_orders_user_status_created ON orders(user_id, status, created_at DESC); ``` ### Step 3: Schema Design Principles **When to normalize (separate tables):** - Data that changes independently (user profile vs order history) - Data referenced from multiple places (products in multiple orders) - Data you query or filter on directly **When to denormalize (JSONB, embedded):** - Snapshot data that shouldn't change when source changes (shipping address on order) - Rarely queried nested data (user preferences blob) - Write-heavy analytics data **Index decision checklist:** - Every foreign key gets an index (PostgreSQL doesn't auto-create these) - Every column in a WHERE clause that filters more than 10% of rows - Every column used in ORDER BY on paginated queries - Composite indexes: put equality columns first, range/sort columns last **STOPPING POINT 3**: Your schema is designed. What next? 1. **Write migration files** - Generate up/down migrations with a migration tool 2. **Add soft deletes** - Track deleted records without losing data 3. **Plan for audit logging** - Track who changed what and when 4. **Optimize for read patterns** - Add materialized views or read replicas --- ## Workflow 3: Design System Architecture for Scale ### Step 1: Start with the Monolith Unless you have proven need, start with a well-structured monolith: ``` my-service/ src/ api/ # HTTP handlers, request/response types services/ # Business logic (one file per domain) repositories/ # Database access (one per table/aggregate) jobs/ # Background job processors events/ # Event publishers and handlers middleware/ # Auth, logging, rate limiting migrations/ # Database migrations config/ # Environment-specific config ``` This structure lets you extract services later by pulling out a `services/` + `repositories/` pair. ### Step 2: Scaling Decision Tree ``` Is your system slow? ├── Yes, the database is the bottleneck │ ├── Read-heavy? → Add read replicas + connection pooling (PgBouncer) │ ├── Write-heavy? → Partition tables, batch writes, use a queue │ └── Single slow query? → Add indexes, rewrite query, add caching ├── Yes, the application server is the bottleneck │ ├── CPU-bound? → Horizontal scale (more instances behind load balancer) │ ├── Memory-bound? → Check for leaks, reduce per-request allocation │ └── I/O-bound? → Use async/await, connection pooling ├── Yes, an external service is slow │ ├── Can you cache responses? → Cache with TTL │ ├── Can you do it async? → Queue + background worker │ └── Neither? → Circuit breaker + timeout + fallback └── No, but I expect growth └── Focus on: connection pooling, index coverage, horizontal app scaling ``` ### Step 3: Caching Strategy Layer your caches: | Layer | Tool | TTL | Use For | |-------|------|-----|---------| | Application | In-memory (LRU) | 1-5 min | Config, feature flags, hot lookups | | Distributed | Redis/Memcached | 5-60 min | Session data, API responses, computed results | | Database | Materialized views | Refresh on schedule | Dashboard queries, aggregations | | CDN | Cloudflare/CloudFront | 1-24 hours | Static assets, public API responses | ```python import redis import json import hashlib cache = redis.Redis(host="localhost", port=6379, decode_responses=True) def cached(ttl_seconds: int = 300): def decorator(fn): def wrapper(*args, **kwargs): key = f"{fn.__name__}:{hashlib.md5(json.dumps([args, kwargs], sort_keys=True, default=str).encode()).hexdigest()}" cached_result = cache.get(key) if cached_result: return json.loads(cached_result) result = fn(*args, **kwargs) cache.setex(key, ttl_seconds, json.dumps(result, default=str)) return result return wrapper return decorator ``` **STOPPING POINT 4**: Architecture is designed. What next? 1. **Add a message queue** - Decouple services with async event processing 2. **Plan for failure** - Circuit breakers, retries, dead letter queues 3. **Extract a microservice** - Pull one bounded context into its own service 4. **Add observability** - Structured logging, metrics, distributed tracing --- ## Workflow 4: Plan a Migration Strategy ### Step 1: Assess the Migration **Migration risk matrix:** | What's Changing | Risk Level | Strategy | |----------------|------------|----------| | Add new column (nullable) | Low | Single migration, no downtime | | Add new column (NOT NULL) | Medium | Add nullable, backfill, then add constraint | | Rename column | High | New column, dual-write, backfill, swap reads, drop old | | Change column type | High | New column approach, same as rename | | Split a table | Very High | New tables, dual-write, migrate reads, drop old | | New database entirely | Very High | Strangler fig pattern (below) | ### Step 2: Zero-Downtime Column Migration ```sql -- Step 1: Add new column (nullable, no lock) ALTER TABLE users ADD COLUMN display_name VARCHAR(255); -- Step 2: Backfill in batches (application code) -- Don't do UPDATE users SET display_name = name; -- locks entire table ``` ```python # Backfill in batches BATCH_SIZE = 1000 while True: rows = db.execute(""" UPDATE users SET display_name = name WHERE id IN ( SELECT id FROM users WHERE display_name IS NULL LIMIT %s ) RETURNING id """, [BATCH_SIZE]) if len(rows) == 0: break time.sleep(0.1) # Don't hammer the database ``` ```sql -- Step 3: After backfill is complete, add the constraint ALTER TABLE users ALTER COLUMN display_name SET NOT NULL; ALTER TABLE users ALTER COLUMN display_name SET DEFAULT ''; ``` ### Step 3: Strangler Fig Pattern (Full System Migration) For migrating from system A to system B: ``` Phase 1: Build system B alongside system A Phase 2: Route NEW writes to system B, continue reading from A Phase 3: Migrate historical data from A to B in background Phase 4: Switch reads to B (with fallback to A) Phase 5: Stop writing to A Phase 6: Decommission A (keep backup for 90 days) ``` ```python class MigrationRouter: def __init__(self, old_service, new_service, feature_flag): self.old = old_service self.new = new_service self.flag = feature_flag async def read(self, id: str): if self.flag.is_enabled("read_from_new"): try: return await self.new.read(id) except NotFoundError: return await self.old.read(id) # Fallback during migration return await self.old.read(id) async def write(self, data: dict): if self.flag.is_enabled("write_to_new"): result = await self.new.write(data) # Dual-write to old system during transition if self.flag.is_enabled("dual_write"): await self.old.write(data) return result return await self.old.write(data) ``` **STOPPING POINT 5**: Your migration plan is ready. What next? 1. **Build a rollback plan** - Define exactly how to revert each phase 2. **Add data validation** - Compare old and new system outputs during migration 3. **Set up monitoring** - Track error rates, latency, and data consistency during cutover 4. **Run a dry run** - Execute the migration against a staging copy of production data --- ## Workflow 5: Set Up Authentication and Authorization ### Step 1: Choose Your Auth Architecture | Approach | Best For | Trade-off | |----------|----------|-----------| | Session-based (cookies) | Server-rendered apps, single domain | Simple but hard to scale across services | | JWT access + refresh tokens | SPAs, mobile apps, microservices | Stateless but can't revoke instantly | | OAuth 2.0 / OIDC (delegated) | "Sign in with Google", enterprise SSO | Standard but complex to implement | | API keys | Server-to-server, developer APIs | Simple but no user context | ### Step 2: JWT Implementation Pattern ```python import jwt from datetime import datetime, timedelta from fastapi import Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials SECRET_KEY = os.environ["JWT_SECRET"] # Use RS256 with key pair in production ALGORITHM = "HS256" ACCESS_TOKEN_TTL = timedelta(minutes=15) REFRESH_TOKEN_TTL = timedelta(days=30) def create_tokens(user_id: str, roles: list[str]) -> dict: now = datetime.utcnow() access_payload = { "sub": user_id, "roles": roles, "type": "access", "iat": now, "exp": now + ACCESS_TOKEN_TTL, } refresh_payload = { "sub": user_id, "type": "refresh", "iat": now, "exp": now + REFRESH_TOKEN_TTL, } return { "access_token": jwt.encode(access_payload, SECRET_KEY, algorithm=ALGORITHM), "refresh_token": jwt.encode(refresh_payload, SECRET_KEY, algorithm=ALGORITHM), "expires_in": int(ACCESS_TOKEN_TTL.total_seconds()), } security = HTTPBearer() async def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), ) -> dict: try: payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM]) if payload.get("type") != "access": raise HTTPException(status_code=401, detail="Invalid token type") return {"user_id": payload["sub"], "roles": payload.get("roles", [])} except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Token expired") except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Invalid token") ``` ### Step 3: Authorization (RBAC) ```python from functools import wraps ROLE_PERMISSIONS = { "admin": ["read", "write", "delete", "manage_users"], "editor": ["read", "write"], "viewer": ["read"], } def require_permission(permission: str): def decorator(fn): @wraps(fn) async def wrapper(*args, current_user: dict = Depends(get_current_user), **kwargs): user_permissions = set() for role in current_user.get("roles", []): user_permissions.update(ROLE_PERMISSIONS.get(role, [])) if permission not in user_permissions: raise HTTPException( status_code=403, detail=f"Permission '{permission}' required", ) return await fn(*args, current_user=current_user, **kwargs) return wrapper return decorator @app.delete("/orders/{order_id}") @require_permission("delete") async def delete_order(order_id: str, current_user: dict = Depends(get_current_user)): await order_service.delete(order_id) return {"status": "deleted"} ``` ### Step 4: Security Checklist Before shipping auth to production: - [ ] Passwords hashed with bcrypt/argon2 (never SHA-256 or MD5) - [ ] JWT secret is at least 256 bits, stored in environment variable - [ ] Access tokens expire in 15 minutes or less - [ ] Refresh tokens stored securely (httpOnly cookie or encrypted storage) - [ ] Rate limiting on login endpoint (max 10 attempts per minute per IP) - [ ] CORS configured to allow only your frontend origins - [ ] All API endpoints require authentication by default (opt-out, not opt-in) - [ ] SQL injection prevented via parameterized queries (never string concatenation) - [ ] Input validation on all user-supplied data - [ ] Sensitive data (tokens, passwords) never logged **STOPPING POINT 6**: Auth is implemented. What next? 1. **Add OAuth/SSO** - Integrate Google, GitHub, or SAML login 2. **Add MFA** - Time-based one-time passwords (TOTP) 3. **Add API keys** - For developer/partner integrations 4. **Add audit logging** - Track every auth event (login, logout, permission change) 5. **Add token revocation** - Blacklist tokens on logout or password change