Database Guide

from fastapi import FastAPI
from svc_infra.api.fastapi.db.sql.add import add_sql_db, setup_sql
from svc_infra.api.fastapi.db.sql.session import SqlSessionDep

app = FastAPI()

# Option 1: Just lifecycle management
add_sql_db(app)  # Reads SQL_URL from environment

# Option 2: Full setup with CRUD routes
from svc_infra.db.sql.resource import SqlResource
from myapp.models import User

setup_sql(
    app,
    resources=[
        SqlResource(model=User, prefix="/users", tags=["users"]),
    ],
)

# Use in routes
@app.get("/users/{user_id}")
async def get_user(user_id: str, session: SqlSessionDep):
    result = await session.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

from fastapi import FastAPI, Depends
from svc_infra.db.nosql.mongo.client import init_mongo, acquire_db, close_mongo
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_mongo()  # Reads MONGO_URL and MONGO_DB from env
    yield
    await close_mongo()

app = FastAPI(lifespan=lifespan)

@app.get("/items")
async def list_items():
    db = await acquire_db()
    items = await db.items.find({}).to_list(100)
    return items

# Example: Compose URL from parts
# .env
DB_DIALECT=postgresql
DB_DRIVER=asyncpg
DB_HOST=localhost
DB_PORT=5432
DB_NAME=myapp
DB_USER=postgres
DB_PASSWORD=secret
DB_PARAMS=sslmode=disable

# Results in: postgresql+asyncpg://postgres:secret@localhost:5432/myapp?sslmode=disable

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    database_url,
    pool_size=20,           # Default: 5
    max_overflow=10,        # Default: 10
    pool_timeout=30,        # Default: 30
    pool_recycle=3600,      # Recycle connections after 1 hour
    pool_pre_ping=True,     # Verify connections before use
)

# PostgreSQL SSL modes (via DB_SSLMODE_DEFAULT or DB_PARAMS)
sslmode=disable     # No SSL (development only!)
sslmode=allow       # Use SSL if available
sslmode=prefer      # Prefer SSL
sslmode=require     # Require SSL (default in svc-infra)
sslmode=verify-ca   # Require SSL + verify CA
sslmode=verify-full # Require SSL + verify CA + hostname

# Automatic for PostgreSQL URLs when DB_SSLMODE_DEFAULT is set
# or when URL contains sslmode=require

# Using CLI
svc-infra sql init

# Or programmatically
from svc_infra.db.sql.core import init_alembic
init_alembic(script_location="migrations")

# Auto-generate from model changes
svc-infra sql revision --autogenerate -m "add users table"

# Empty migration for manual edits
svc-infra sql revision -m "seed data"

# Programmatic
from svc_infra.db.sql.core import revision

revision(
    message="add users table",
    autogenerate=True,
    ensure_head_before_autogenerate=True,
)

# Upgrade to latest
svc-infra sql upgrade head

# Upgrade to specific revision
svc-infra sql upgrade abc123

# Downgrade one step
svc-infra sql downgrade -1

# Downgrade to specific revision
svc-infra sql downgrade abc123

# Programmatic
from svc_infra.db.sql.core import upgrade, downgrade

upgrade()              # To head
upgrade("abc123")      # To specific revision
downgrade(steps=1)     # One step back

# Automatic: imports ModelBase.metadata from svc_infra.db.sql.base
from svc_infra.db.sql.base import ModelBase

class User(ModelBase):
    __tablename__ = "users"
    id = Column(String, primary_key=True)
    email = Column(String, unique=True)

# Models extending ModelBase are auto-discovered

# alembic.ini can point to different databases per environment
# Use ALEMBIC_CONFIG env var to switch configurations

# CI/CD example:
# 1. Test migrations against ephemeral DB
# 2. Validate with --sql flag
svc-infra sql upgrade head --sql > migration.sql

from svc_infra.api.fastapi.db.sql.session import SqlSessionDep

@app.post("/users")
async def create_user(data: UserCreate, session: SqlSessionDep):
    user = User(**data.dict())
    session.add(user)
    # Auto-commits on success, rolls back on exception
    return user

@app.post("/transfer")
async def transfer(from_id: str, to_id: str, amount: int, session: SqlSessionDep):
    # Both operations in same transaction
    from_account = await session.get(Account, from_id)
    from_account.balance -= amount

    to_account = await session.get(Account, to_id)
    to_account.balance += amount

    # Commits together or rolls back together
    return {"status": "ok"}

# PostgreSQL only
DB_STATEMENT_TIMEOUT_MS=30000  # 30 seconds

# Applied automatically via SET LOCAL statement_timeout
# Scoped to the current transaction only

from svc_infra.db.sql.repository import SqlRepository

repo = SqlRepository(
    model=User,
    id_attr="id",
    soft_delete=True,
    soft_delete_field="deleted_at",
    immutable_fields={"id", "created_at"},
)

# List with pagination
users = await repo.list(session, limit=10, offset=0, order_by=[User.created_at.desc()])

# Get by ID (respects soft-delete)
user = await repo.get(session, "user-123")

# Create
new_user = await repo.create(session, {"email": "test@example.com"})

# Update (ignores immutable fields)
updated = await repo.update(session, "user-123", {"name": "New Name"})

# Delete (soft or hard based on configuration)
await repo.delete(session, "user-123")

# Search with ILIKE
results = await repo.search(
    session,
    q="john",
    fields=["name", "email"],
    limit=10,
    offset=0,
)

from svc_infra.db.sql.service import SqlService

class UserService(SqlService):
    async def pre_create(self, data: dict) -> dict:
        # Hash password, validate email, etc.
        data["password_hash"] = hash_password(data.pop("password"))
        return data

    async def pre_update(self, data: dict) -> dict:
        # Audit logging, validation
        return data

# Usage
service = UserService(repo)
user = await service.create(session, {"email": "...", "password": "..."})

from svc_infra.db.sql.tenant import TenantSqlService

# All operations scoped to tenant_id
service = TenantSqlService(
    repo,
    tenant_id="tenant-123",
    tenant_field="tenant_id",
)

# List only returns this tenant's records
users = await service.list(session, limit=10, offset=0)

# Create auto-injects tenant_id
user = await service.create(session, {"email": "..."})
# user.tenant_id == "tenant-123"

from svc_infra.db.nosql.mongo.client import init_mongo, acquire_db, close_mongo

# Initialize with custom settings
from svc_infra.db.nosql.mongo.settings import MongoSettings

settings = MongoSettings(
    url="mongodb://user:pass@cluster.mongodb.net",
    db_name="myapp",
    max_pool_size=50,
)
db = await init_mongo(settings)

# Get database reference
db = await acquire_db()

# Operations
await db.users.insert_one({"email": "test@example.com"})
users = await db.users.find({}).to_list(100)

# Cleanup
await close_mongo()

from svc_infra.db.nosql.mongo.client import ping_mongo

is_healthy = await ping_mongo()  # Returns True if connected

from svc_infra.api.fastapi.db.sql.add import add_sql_health

add_sql_health(app, prefix="/_sql/health")
# Exposes GET /_sql/health with connection status

from svc_infra.health import HealthRegistry, check_database, check_redis

registry = HealthRegistry()
registry.add("database", check_database(os.getenv("SQL_URL")), critical=True)
registry.add("redis", check_redis(os.getenv("REDIS_URL")), critical=False)

# Wait for dependencies at startup
await registry.wait_until_healthy(timeout=60, interval=2)

from svc_infra.db.sql.resource import SqlResource

resources = [
    SqlResource(
        model=User,
        prefix="/users",
        tags=["users"],
        id_attr="id",
        soft_delete=True,
        search_fields=["name", "email"],
        ordering_default="-created_at",
        allowed_order_fields=["created_at", "name", "email"],
        # Tenant scoping
        tenant_field="tenant_id",
        # Custom service
        service_factory=lambda repo: UserService(repo),
    ),
]

setup_sql(app, resources)

# Auto-generates:
# GET    /users         - List with pagination, search, ordering
# POST   /users         - Create
# GET    /users/{id}    - Get by ID
# PATCH  /users/{id}    - Update
# DELETE /users/{id}    - Delete (soft if configured)

# PostgreSQL: connections = (cores * 2) + spindles
# Rule of thumb: pool_size = 10-20 per worker

# For 4 Gunicorn workers with 10 threads each:
# pool_size = 5, max_overflow = 10 per worker
# Total: 4 * 15 = 60 max connections

# Use different URLs for read vs write
WRITE_DB_URL = os.getenv("PRIMARY_DB_URL")
READ_DB_URL = os.getenv("REPLICA_DB_URL", WRITE_DB_URL)

# Create separate engines
write_engine = create_async_engine(WRITE_DB_URL)
read_engine = create_async_engine(READ_DB_URL)

from svc_infra.data.lifecycle import verify_backups

# Check backup health
report = await verify_backups(
    expected_max_age_hours=24,
    backup_path="/backups",
)
if not report.healthy:
    alert_ops_team(report)

# API endpoints: 30 seconds
DB_STATEMENT_TIMEOUT_MS=30000

# Background jobs: 5 minutes
# Set per-job via session.execute(text("SET LOCAL statement_timeout = 300000"))

sqlalchemy.exc.OperationalError: connection refused

# Enable SQLAlchemy logging
import logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

# Set statement timeout to catch runaway queries
DB_STATEMENT_TIMEOUT_MS=30000

alembic.util.exc.CommandError: Multiple heads

svc-infra sql merge heads -m "merge branches"

TimeoutError: QueuePool limit of size X overflow Y reached

motor.motor_asyncio.AsyncIOMotorClient: Connection refused