tomevault-io FastAPI + Dapr Backend
Build production-grade FastAPI backends with SQLModel, Dapr integration, and JWT authentication.
Quick Start
# Project setup
uv init backend && cd backend
uv add fastapi sqlmodel pydantic httpx python-jose uvicorn
# Development
uv run uvicorn main:app --reload --port 8000
# With Dapr sidecar
dapr run --app-id myapp --app-port 8000 -- uvicorn main:app
FastAPI Core Patterns
1. SQLModel Schema (Database + API)
from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional, Literal
class TaskBase(SQLModel):
title: str = Field(max_length=200, index=True)
status: Literal["pending", "in_progress", "completed"] = "pending"
class Task(TaskBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
created_at: datetime = Field(default_factory=datetime.now)
class TaskCreate(TaskBase):
pass
class TaskRead(TaskBase):
id: int
created_at: datetime
2. Async Database Setup
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
import os
DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
engine = create_async_engine(DATABASE_URL)
async def get_session() -> AsyncSession:
async with AsyncSession(engine) as session:
yield session
3. CRUD Endpoints
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import select
app = FastAPI()
@app.post("/tasks", response_model=TaskRead, status_code=201)
async def create_task(task: TaskCreate, session: AsyncSession = Depends(get_session)):
db_task = Task.model_validate(task)
session.add(db_task)
await session.commit()
await session.refresh(db_task)
return db_task
@app.get("/tasks/{task_id}", response_model=TaskRead)
async def get_task(task_id: int, session: AsyncSession = Depends(get_session)):
task = await session.get(Task, task_id)
if not task:
raise HTTPException(status_code=404, detail="Not found")
return task
@app.patch("/tasks/{task_id}", response_model=TaskRead)
async def update_task(task_id: int, update: TaskUpdate, session: AsyncSession = Depends(get_session)):
task = await session.get(Task, task_id)
if not task:
raise HTTPException(status_code=404, detail="Not found")
update_data = update.model_dump(exclude_unset=True)
task.sqlmodel_update(update_data)
session.add(task)
await session.commit()
await session.refresh(task)
return task
4. JWT/JWKS Authentication
from jose import jwt
import httpx
JWKS_URL = f"{SSO_URL}/.well-known/jwks.json"
async def get_current_user(authorization: str = Header()):
token = authorization.replace("Bearer ", "")
async with httpx.AsyncClient() as client:
jwks = (await client.get(JWKS_URL)).json()
payload = jwt.decode(token, jwks, algorithms=["RS256"])
return payload
@app.get("/protected")
async def protected_route(user = Depends(get_current_user)):
return {"user": user["sub"]}
See references/fastapi-patterns.md for audit logging, pagination, and OpenAPI configuration.
Dapr Integration Patterns
1. Pub/Sub Subscription
from fastapi import APIRouter, Request
router = APIRouter(prefix="/dapr", tags=["Dapr"])
@router.get("/subscribe")
async def subscribe():
"""Dapr calls this to discover subscriptions."""
return [{
"pubsubname": "pubsub",
"topic": "task-created",
"route": "/dapr/task-created"
}]
@router.post("/task-created")
async def handle_task_created(request: Request, session: AsyncSession = Depends(get_session)):
# CloudEvent wrapper - data is nested
event = await request.json()
task_data = event.get("data", event) # Handle both wrapped and unwrapped
# Process event
task = Task.model_validate(task_data)
session.add(task)
await session.commit()
return {"status": "processed"}
2. Publishing Events
import httpx
DAPR_URL = "http://localhost:3500"
async def publish_event(topic: str, data: dict):
async with httpx.AsyncClient() as client:
await client.post(
f"{DAPR_URL}/v1.0/publish/pubsub/{topic}",
json=data,
headers={"Content-Type": "application/json"}
)
3. Scheduled Jobs
# Schedule a job via Dapr Jobs API (alpha)
async def schedule_job(name: str, schedule: str, callback_url: str, data: dict):
async with httpx.AsyncClient() as client:
await client.post(
f"{DAPR_URL}/v1.0-alpha1/jobs/{name}",
json={
"schedule": schedule, # "@every 5m" or "0 */5 * * * *"
"data": data,
},
headers={"dapr-app-callback-url": callback_url}
)
# Job callback endpoint
@app.post("/jobs/process")
async def process_job(request: Request):
job_data = await request.json()
# Handle job execution
return {"status": "completed"}
See references/dapr-patterns.md for state management and advanced patterns.
Production Patterns
Structured Logging
import structlog
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer()
]
)
log = structlog.get_logger()
log.info("task_created", task_id=task.id, user_id=user["sub"])
Repository + Service Pattern
# Repository: data access only
class TaskRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def create(self, task: TaskCreate) -> Task:
db_task = Task.model_validate(task)
self.session.add(db_task)
await self.session.commit()
return db_task
# Service: business logic
class TaskService:
def __init__(self, repo: TaskRepository):
self.repo = repo
async def create_task(self, task: TaskCreate, user_id: str) -> Task:
# Business logic here
return await self.repo.create(task)
# Dependency injection
def get_task_service(session: AsyncSession = Depends(get_session)):
return TaskService(TaskRepository(session))
Async Testing
@pytest.fixture
async def client(session):
app.dependency_overrides[get_session] = lambda: session
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as ac:
yield ac
@pytest.mark.anyio
async def test_create_task(client: AsyncClient):
response = await client.post("/tasks", json={"title": "Test"})
assert response.status_code == 201
See references/production-testing.md for full patterns.
Project Structure
backend/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app
│ ├── database.py # Async engine + session
│ ├── models/ # SQLModel schemas
│ ├── routers/ # API routes
│ ├── repositories/ # Data access layer
│ ├── services/ # Business logic
│ └── dapr/ # Dapr handlers
├── tests/
│ ├── conftest.py # Fixtures
│ └── test_*.py # Test files
├── components/ # Dapr components (k8s)
│ ├── pubsub.yaml
│ └── statestore.yaml
└── pyproject.toml
Verification
Run: python3 scripts/verify.py
Expected: ✓ scaffolding-fastapi-dapr skill ready
If Verification Fails
- Check: references/ folder has both pattern files
- Stop and report if still failing
Related Skills
- configuring-better-auth - JWT/JWKS auth for API endpoints
- fetching-library-docs - FastAPI docs:
--library-id /fastapi/fastapi --topic dependencies
References
- references/fastapi-patterns.md - Complete FastAPI backend patterns
- references/dapr-patterns.md - Dapr pub/sub, state, and jobs
- references/sqlmodel-patterns.md - SQLModel database patterns and migrations
- references/production-testing.md - Structured logging, DI, testing, versioning
Converted and distributed by TomeVault — claim your Tome and manage your conversions.