api

api

Production patterns for building and consuming APIs in the Python/FastAPI + TypeScript/React stack used across lead-gen-engine and seo-geo-aeo-engine.

1) FastAPI App Structure

# api/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: init DB pool, Redis, scheduler
    await db.connect()
    yield
    # Shutdown: close pool
    await db.disconnect()

app = FastAPI(lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://localhost:3002"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Register routers
from api.routes import auth, campaigns, prospects
app.include_router(auth.router, prefix="/api/auth")
app.include_router(campaigns.router, prefix="/api/campaigns")
app.include_router(prospects.router, prefix="/api")

2) Route Module Pattern

# api/routes/campaigns.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from api.deps import get_db, get_current_user

router = APIRouter()

class OnboardRequest(BaseModel):
    name: str
    domain: str
    target_icp: str | None = None

class CampaignResponse(BaseModel):
    id: int
    name: str
    status: str

@router.post("/campaigns/onboard", response_model=CampaignResponse)
def onboard_campaign(body: OnboardRequest, db=Depends(get_db)):
    campaign = db.insert("campaigns", {"name": body.name, "domain": body.domain})
    return campaign

@router.get("/campaigns", response_model=list[CampaignResponse])
def list_campaigns(status: str | None = None, db=Depends(get_db)):
    filters = {"status": status} if status else {}
    return db.fetch_many("campaigns", filters)

3) JWT Auth Pattern

# shared/auth.py
import os
from datetime import datetime, timedelta
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = os.environ.get("JWT_SECRET_KEY", "change-me-in-production")
ALGORITHM = "HS256"
EXPIRE_MINUTES = int(os.environ.get("JWT_EXPIRE_MINUTES", "480"))

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

def create_access_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + timedelta(minutes=EXPIRE_MINUTES)
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return {"id": payload["sub"], "email": payload["email"], "role": payload["role"]}
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)