⚙️ Backend Development Strategy
Backend architecture should be chosen based on discovery phase findings about scale, real-time needs, and team expertise
Supabase
Best for: Rapid development, real-time features, built-in auth
FastAPI
Best for: Python teams, ML integration, high performance
LangGraph
Best for: AI workflows, stateful processes, agent systems
Pre-Implementation Research Process
Always research before implementing. Use this two-step process for optimal results
Research Phase with Codex/Sonnet
I need to implement [FEATURE] with these requirements:
- [Requirement from discovery]
- Scale: [Expected load from discovery]
- Real-time: [Yes/No from discovery]
Current stack: [Your chosen stack]
Please suggest:
1. Database schema design
2. API structure
3. Caching strategy
4. Security considerations
5. Performance optimizations
6. Edge cases to handle
Implementation with Claude Code
Based on this architecture review from [Codex/Sonnet]:
[Paste research findings]
And these discovery requirements:
- [Specific requirement]
- [Performance target]
- [Security requirement]
Now implement:
1. Database migrations
2. RLS policies (if Supabase)
3. API endpoints with validation
4. Error handling
5. Client-side hooks
6. Tests
Follow our established patterns from [reference files]
Backend Architecture Patterns
- Supabase Architecture
- FastAPI Architecture
- LangGraph Workflows
Complete Supabase Setup
Ideal for most SaaS applications with real-time needs
Database Design
-- From discovery data requirements
-- Users and profiles
CREATE TABLE profiles (
id UUID PRIMARY KEY REFERENCES auth.users(id),
email TEXT UNIQUE NOT NULL,
full_name TEXT,
avatar_url TEXT,
role user_role NOT NULL DEFAULT 'user',
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- Enable RLS
ALTER TABLE profiles ENABLE ROW LEVEL SECURITY;
-- Policies based on discovery requirements
CREATE POLICY "Users can view own profile"
ON profiles FOR SELECT
USING (auth.uid() = id);
CREATE POLICY "Users can update own profile"
ON profiles FOR UPDATE
USING (auth.uid() = id);
-- Indexes for performance
CREATE INDEX idx_profiles_email ON profiles(email);
CREATE INDEX idx_profiles_role ON profiles(role);
Edge Functions
// supabase/functions/process-payment/index.ts
import { serve } from 'https://deno.land/std@0.168.0/http/server.ts'
import { createClient } from '@supabase/supabase-js'
serve(async (req) => {
try {
const supabase = createClient(
Deno.env.get('SUPABASE_URL')!,
Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
)
const { amount, userId } = await req.json()
// Process payment logic
const result = await processPayment(amount)
// Update database
await supabase
.from('payments')
.insert({ user_id: userId, amount, status: 'completed' })
return new Response(JSON.stringify(result), {
headers: { 'Content-Type': 'application/json' },
})
} catch (error) {
return new Response(JSON.stringify({ error: error.message }), {
status: 400,
headers: { 'Content-Type': 'application/json' },
})
}
})
Real-time Subscriptions
// Real-time features from discovery
const channel = supabase
.channel('room-1')
.on('postgres_changes', {
event: '*',
schema: 'public',
table: 'messages'
}, (payload) => {
handleNewMessage(payload.new)
})
.on('presence', { event: 'sync' }, () => {
const state = channel.presenceState()
updateOnlineUsers(state)
})
.subscribe()
Python FastAPI Setup
Ideal for ML integration and Python teams
# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from app.core.config import settings
from app.core.security import get_current_user
from app.db.session import get_db
from app.api import auth, users, items
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.VERSION,
openapi_url="/api/openapi.json"
)
# CORS from discovery requirements
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])
app.include_router(items.router, prefix="/api/items", tags=["items"])
@app.get("/health")
async def health_check():
return {"status": "healthy"}
# Background tasks
from app.workers import celery_app
@app.post("/api/tasks/process")
async def create_task(
data: dict,
current_user = Depends(get_current_user),
db: Session = Depends(get_db)
):
task = celery_app.send_task(
"process_data",
args=[data, current_user.id]
)
return {"task_id": task.id}
AI Workflow Architecture
For complex AI-powered features from discovery
# graphs/customer_support.py
from langgraph.graph import StateGraph, State
from langchain.chat_models import ChatOpenAI
from typing import TypedDict, List
class ConversationState(TypedDict):
messages: List[str]
current_intent: str
resolved: bool
escalate: bool
# Define the graph
workflow = StateGraph(ConversationState)
# Add nodes
workflow.add_node("classify_intent", classify_intent_node)
workflow.add_node("handle_faq", handle_faq_node)
workflow.add_node("collect_info", collect_info_node)
workflow.add_node("escalate_human", escalate_human_node)
workflow.add_node("resolve", resolve_node)
# Add edges
workflow.add_edge("classify_intent", route_by_intent)
workflow.add_conditional_edges(
"handle_faq",
check_if_resolved,
{
True: "resolve",
False: "collect_info"
}
)
# Compile
app = workflow.compile()
# Use in API
@app.post("/api/support/chat")
async def chat(message: str, session_id: str):
result = await app.ainvoke({
"messages": [message],
"current_intent": None,
"resolved": False,
"escalate": False
})
return result
Next Step
With backend architecture defined, proceed to testing strategies →