Spaces:
Sleeping
Sleeping
FastAPI Developer Markdown Brain
Used by AI pipeline to generate production FastAPI backends with streaming, auth, and real integrations.
Project Structure
backend/
├── app.py # OR main.py — FastAPI entrypoint
├── requirements.txt
├── .env
├── Dockerfile
├── routers/
│ ├── __init__.py
│ ├── auth.py
│ ├── posts.py
│ └── ai.py
├── models/
│ ├── __init__.py
│ ├── user.py
│ └── post.py
├── schemas/
│ ├── __init__.py
│ └── post.py
├── services/
│ ├── __init__.py
│ ├── ai_service.py
│ └── auth_service.py
├── db/
│ ├── __init__.py
│ └── database.py
└── tests/
├── conftest.py
└── test_posts.py
FastAPI App Entrypoint
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from routers import auth, posts, ai
import os
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
print("Starting up...")
yield
# Shutdown
print("Shutting down...")
app = FastAPI(
title="InStatic CMS API",
description="AI-powered content management and builder API",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Tighten in production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth.router, prefix="/auth", tags=["auth"])
app.include_router(posts.router, prefix="/posts", tags=["posts"])
app.include_router(ai.router, prefix="/ai", tags=["ai"])
@app.get("/health")
async def health():
return {"status": "ok", "service": "instatic-cms"}
JWT Authentication
# services/auth_service.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
SECRET_KEY = os.getenv("SECRET_KEY", "change-me-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # 24 hours
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
return user_id
Streaming Response (SSE)
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
import httpx, json, asyncio
router = APIRouter()
@router.post("/stream")
async def stream_ai(request: dict):
async def generate():
async with httpx.AsyncClient(timeout=120) as client:
async with client.stream(
"POST",
"https://api.groq.com/openai/v1/chat/completions",
headers={"Authorization": f"Bearer {os.getenv('GROQ_API_KEY')}"},
json={
"model": "llama-3.3-70b-versatile",
"messages": request.get("messages", []),
"stream": True,
"max_tokens": 4096,
},
) as response:
async for line in response.aiter_lines():
if line.startswith("data: "):
chunk = line[6:]
if chunk == "[DONE]":
yield "data: [DONE]\n\n"
break
try:
data = json.loads(chunk)
content = data["choices"][0]["delta"].get("content", "")
if content:
yield f"data: {json.dumps({'content': content})}\n\n"
except Exception:
pass
return StreamingResponse(generate(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
LLM Fallback Chain (Groq → Anthropic → OpenRouter → Cerebras)
import httpx, os, asyncio
PROVIDERS = [
{
"name": "groq",
"url": "https://api.groq.com/openai/v1/chat/completions",
"key_env": "GROQ_API_KEY",
"model": "llama-3.3-70b-versatile",
},
{
"name": "anthropic",
"url": "https://api.anthropic.com/v1/messages",
"key_env": "ANTHROPIC_API_KEY",
"model": "claude-sonnet-4-6",
},
{
"name": "openrouter",
"url": "https://openrouter.ai/api/v1/chat/completions",
"key_env": "OPENROUTER_API_KEY",
"model": "meta-llama/llama-3.3-70b-instruct",
},
{
"name": "cerebras",
"url": "https://api.cerebras.ai/v1/chat/completions",
"key_env": "CEREBRAS_API_KEY",
"model": "llama3.1-70b",
},
]
async def call_llm_with_fallback(messages: list, max_tokens: int = 4096) -> str:
for provider in PROVIDERS:
key = os.getenv(provider["key_env"])
if not key:
continue
try:
async with httpx.AsyncClient(timeout=60) as client:
if provider["name"] == "anthropic":
resp = await client.post(
provider["url"],
headers={"x-api-key": key, "anthropic-version": "2023-06-01"},
json={"model": provider["model"], "max_tokens": max_tokens,
"messages": messages},
)
resp.raise_for_status()
return resp.json()["content"][0]["text"]
else:
resp = await client.post(
provider["url"],
headers={"Authorization": f"Bearer {key}"},
json={"model": provider["model"], "max_tokens": max_tokens,
"messages": messages},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"]
except Exception as e:
print(f"[{provider['name']}] failed: {e}, trying next...")
continue
raise RuntimeError("All LLM providers failed")
SQLite with aiosqlite
# db/database.py
import aiosqlite, os
DB_PATH = os.getenv("DB_PATH", "cms.db")
async def get_db():
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
yield db
async def init_db():
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS posts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
content TEXT NOT NULL,
status TEXT DEFAULT 'draft',
created TEXT DEFAULT (datetime('now')),
updated TEXT DEFAULT (datetime('now'))
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS builds (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prompt TEXT NOT NULL,
status TEXT DEFAULT 'pending',
result TEXT,
error TEXT,
created TEXT DEFAULT (datetime('now'))
)
""")
await db.commit()
Pydantic Schemas
from pydantic import BaseModel, Field
from typing import Optional, Literal
from datetime import datetime
class PostCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
content: str = Field(..., min_length=1)
status: Literal["draft", "publish"] = "draft"
class PostResponse(BaseModel):
id: int
title: str
content: str
status: str
created: datetime
updated: datetime
class Config:
from_attributes = True
class BuildRequest(BaseModel):
prompt: str = Field(..., min_length=10, max_length=4000)
target: Literal["website", "android", "api", "wordpress"] = "website"
style: Optional[str] = None
class BuildResponse(BaseModel):
id: int
status: str
result: Optional[dict] = None
error: Optional[str] = None
requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.30.6
httpx==0.27.0
pydantic==2.8.2
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
aiosqlite==0.20.0
python-multipart==0.0.9
python-dotenv==1.0.1
groq==0.9.0
anthropic==0.34.0
Dockerfile (HuggingFace Space)
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy source
COPY . .
# HF Spaces runs on port 7860
EXPOSE 7860
# Start server
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1"]