Skip to content

Database Standards

Operational guides for working with PostgreSQL in our stack — migrations, connection management, monitoring, and implementation patterns.

For theory and deep knowledge (SQL fundamentals, query performance, indexing, transactions, backup), see Database Internals.

Overview

PostgreSQL is our primary database. We use it for:

  • Relational data — Users, sessions, bookings, media metadata
  • JSONB storage — Flexible schemas for configuration, preferences
  • Full-text search — Native search without external services
  • Geospatial queries — PostGIS for location-based features

Architecture

┌─────────────────┐     ┌──────────────────┐     ┌─────────────┐
│   FastAPI App   │────▶│    SQLAlchemy    │────▶│  PostgreSQL │
│   (async)       │     │    (asyncpg)     │     │    15+      │
└─────────────────┘     └──────────────────┘     └─────────────┘
                              │
                              ▼
                        ┌──────────────────┐
                        │     Alembic      │
                        │   (migrations)   │
                        └──────────────────┘

When to Use What

Need Solution
Simple queries SQLAlchemy ORM
Complex aggregations Raw SQL via text()
Bulk inserts COPY or executemany
Search PostgreSQL full-text search
JSON data JSONB columns
Geospatial PostGIS extension

Section Contents

Operations

Patterns

Reference

  • Cheatsheet — Quick reference for common operations

Quick Start

Connect to Local Database

# Using psql
psql -h localhost -U postgres -d sartiq

# Using pgcli (better autocomplete)
pgcli -h localhost -U postgres -d sartiq

Run Migrations

# Apply all pending migrations
uv run alembic upgrade head

# Create new migration
uv run alembic revision --autogenerate -m "Add user preferences table"

# Rollback one migration
uv run alembic downgrade -1

Common Query Patterns

from sqlalchemy import select, func
from app.models import User, Booking

# Async query with SQLAlchemy
async def get_user_with_bookings(db: AsyncSession, user_id: int):
    stmt = (
        select(User)
        .options(selectinload(User.bookings))
        .where(User.id == user_id)
    )
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

# Aggregation
async def get_booking_stats(db: AsyncSession):
    stmt = (
        select(
            func.count(Booking.id).label("total"),
            func.sum(Booking.amount).label("revenue")
        )
        .where(Booking.status == "completed")
    )
    result = await db.execute(stmt)
    return result.one()

Best Practices

  1. Always use parameterized queries — Never interpolate user input
  2. Index foreign keys — PostgreSQL doesn't auto-index them
  3. Use EXPLAIN before optimizing — Measure, don't guess
  4. Keep transactions short — Long transactions block others
  5. Prefer bulk operations — Batch inserts/updates when possible