feat: add auth infrastructure, DB sessions, and S3 helpers

Backport battle-tested patterns from WIT into the app template so new
apps start with JWT auth (register/login/me), User model, database
session management, S3 storage helpers, CORS middleware, and a proper
test infrastructure with SQLite fixtures.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: baijum

README.md

@@ -14,7 +14,14 @@ A template repository for bootstrapping new applications on the [Towlion platfor
 app/                    # FastAPI backend
   main.py               # Application entry point
   Dockerfile            # Backend container image
-  models.py             # SQLAlchemy models
+  models.py             # SQLAlchemy models (User)
+  database.py           # DB session management
+  auth.py               # JWT + bcrypt helpers
+  deps.py               # FastAPI dependencies (get_current_user)
+  schemas.py            # Pydantic request/response models
+  storage.py            # S3/MinIO file helpers
+  routers/
+    auth.py             # Register, login, profile endpoints
   alembic/              # Database migrations
 deploy/
   docker-compose.yml    # App containers (multi-app mode)
@@ -39,6 +46,35 @@ uvicorn app.main:app --reload --port 8000
 curl http://localhost:8000/health
 ```
 
+## Authentication
+
+The template includes JWT-based authentication out of the box:
+
+| Endpoint | Method | Description |
+|---|---|---|
+| `/api/auth/register` | POST | Create account (email, display_name, password) |
+| `/api/auth/login` | POST | Get token (email, password) |
+| `/api/auth/me` | GET | Current user profile (requires token) |
+
+Use the `Authorization: Bearer <token>` header for authenticated requests.
+
+To protect your own endpoints, add the `get_current_user` dependency:
+
+```python
+from fastapi import Depends
+from app.deps import get_current_user
+from app.models import User
+
+@app.get("/api/my-endpoint")
+def my_endpoint(user: User = Depends(get_current_user)):
+    return {"user_id": user.id}
+```
+
+**Environment variables:**
+- `JWT_SECRET` — Secret key for signing tokens (required in production)
+- `JWT_EXPIRY_HOURS` — Token lifetime, default `72`
+- `CORS_ORIGINS` — Comma-separated allowed origins, default `http://localhost:3000`
+
 ## Environment Variables
 
 Copy `deploy/env.template` to `deploy/.env` and fill in your values. See the [platform spec](https://github.com/towlion/platform/blob/main/docs/spec.md) for details on required and optional variables.

app/alembic/versions/0001_users.py

@@ -0,0 +1,33 @@
+"""create users table
+
+Revision ID: 0001
+Revises:
+Create Date: 2026-03-17
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0001"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "users",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("email", sa.String(255), unique=True, nullable=False),
+        sa.Column("display_name", sa.String(255), nullable=False),
+        sa.Column("password_hash", sa.String(255), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+        ),
+    )
+
+
+def downgrade():
+    op.drop_table("users")

app/auth.py

@@ -0,0 +1,33 @@
+import os
+from datetime import datetime, timedelta, timezone
+
+import bcrypt
+import jwt
+
+JWT_SECRET = os.getenv("JWT_SECRET", "dev-secret-change-me")
+JWT_ALGORITHM = "HS256"
+JWT_EXPIRY_HOURS = int(os.getenv("JWT_EXPIRY_HOURS", "72"))
+
+
+def hash_password(password: str) -> str:
+    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    return bcrypt.checkpw(plain.encode(), hashed.encode())
+
+
+def create_token(user_id: int) -> str:
+    payload = {
+        "sub": str(user_id),
+        "exp": datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRY_HOURS),
+    }
+    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
+
+
+def decode_token(token: str) -> int | None:
+    try:
+        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
+        return int(payload["sub"])
+    except (jwt.PyJWTError, KeyError, ValueError):
+        return None

app/database.py

@@ -0,0 +1,18 @@
+import os
+from collections.abc import Generator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://localhost:5432/myapp")
+
+engine = create_engine(DATABASE_URL)
+SessionLocal = sessionmaker(bind=engine)
+
+
+def get_db() -> Generator[Session, None, None]:
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/deps.py

@@ -0,0 +1,22 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy.orm import Session
+
+from app.auth import decode_token
+from app.database import get_db
+from app.models import User
+
+security = HTTPBearer()
+
+
+def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+    db: Session = Depends(get_db),
+) -> User:
+    user_id = decode_token(credentials.credentials)
+    if user_id is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+    user = db.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
+    return user

app/main.py

@@ -1,14 +1,18 @@
 import logging
+import os
 import sys
 import time
 
 from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from pythonjsonlogger import jsonlogger
 from slowapi import Limiter, _rate_limit_exceeded_handler
 from slowapi.errors import RateLimitExceeded
 from slowapi.util import get_remote_address
 
+from app.routers import auth
+
 # Configure structured JSON logging
 handler = logging.StreamHandler(sys.stdout)
 handler.setFormatter(
@@ -27,6 +31,19 @@
 app.state.limiter = limiter
 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 
+# CORS
+cors_origins = os.getenv("CORS_ORIGINS", "http://localhost:3000").split(",")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=cors_origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Routers
+app.include_router(auth.router, prefix="/api")
+
 
 @app.middleware("http")
 async def log_requests(request: Request, call_next):

app/models.py

@@ -1,8 +1,20 @@
-from sqlalchemy.orm import DeclarativeBase
+import datetime
+
+from sqlalchemy import DateTime, String, func
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
 
 class Base(DeclarativeBase):
     pass
 
 
-# Add your SQLAlchemy models here
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
+    display_name: Mapped[str] = mapped_column(String(255), nullable=False)
+    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
+    created_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )

app/alembic/versions/.gitkeep app/routers/__init__.pyapp/alembic/versions/.gitkeep renamed to app/routers/__init__.py

@@ -0,0 +1,39 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.auth import create_token, hash_password, verify_password
+from app.database import get_db
+from app.deps import get_current_user
+from app.models import User
+from app.schemas import LoginRequest, RegisterRequest, TokenResponse, UserResponse
+
+router = APIRouter(prefix="/auth", tags=["auth"])
+
+
+@router.post("/register", response_model=TokenResponse, status_code=201)
+def register(body: RegisterRequest, db: Session = Depends(get_db)):
+    existing = db.query(User).filter_by(email=body.email).first()
+    if existing:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
+    user = User(
+        email=body.email,
+        display_name=body.display_name,
+        password_hash=hash_password(body.password),
+    )
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return TokenResponse(access_token=create_token(user.id))
+
+
+@router.post("/login", response_model=TokenResponse)
+def login(body: LoginRequest, db: Session = Depends(get_db)):
+    user = db.query(User).filter_by(email=body.email).first()
+    if not user or not verify_password(body.password, user.password_hash):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
+    return TokenResponse(access_token=create_token(user.id))
+
+
+@router.get("/me", response_model=UserResponse)
+def me(user: User = Depends(get_current_user)):
+    return user

app/routers/auth.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+
+from pydantic import BaseModel, EmailStr, Field
+
+
+class RegisterRequest(BaseModel):
+    email: EmailStr = Field(examples=["user@example.com"])
+    display_name: str = Field(examples=["Jane Doe"])
+    password: str = Field(min_length=8)
+
+
+class LoginRequest(BaseModel):
+    email: EmailStr = Field(examples=["user@example.com"])
+    password: str
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+
+
+class UserResponse(BaseModel):
+    id: int
+    email: str
+    display_name: str
+    created_at: datetime
+
+    model_config = {"from_attributes": True}