⚙️ 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
Copy
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
Copy
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
Copy
-- 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
Copy
// supabase/functions/process-payment/index.ts
import { serve } from 'https://deno.land/[email protected]/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
Copy
// 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
Copy
# 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
Copy
# 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 →