database

database

Authoritative reference for relational database work: connection pooling, async queries, migrations, and Supabase-specific patterns.

1) Async connection with asyncpg / SQLAlchemy 2

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host/db",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,   # drop dead connections before use
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

2) SQLAlchemy 2 Core style (preferred for high-throughput)

from sqlalchemy import select, insert, update, delete, text

# SELECT with filter
stmt = select(Article).where(Article.published == True).order_by(Article.created_at.desc()).limit(20)
result = await session.execute(stmt)
rows = result.scalars().all()

# Bulk INSERT (ignore duplicates)
stmt = insert(Keyword).values(data).on_conflict_do_nothing(index_elements=['slug'])
await session.execute(stmt)
await session.commit()

# Raw SQL (last resort — prefer Core)
result = await session.execute(text("SELECT id FROM articles WHERE slug = :s"), {"s": slug})

3) Model definition pattern

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, DateTime, func
import uuid

class Base(DeclarativeBase):
    pass

class Article(Base):
    __tablename__ = "articles"

    id:         Mapped[uuid.UUID]  = mapped_column(primary_key=True, default=uuid.uuid4)
    slug:       Mapped[str]        = mapped_column(String(255), unique=True, index=True)
    title:      Mapped[str]        = mapped_column(String(500))
    published:  Mapped[bool]       = mapped_column(default=False)
    created_at: Mapped[datetime]   = mapped_column(server_default=func.now())

4) Alembic migrations

# Init (once per project)
alembic init alembic

# alembic/env.py — point at your models
from app.models import Base
target_metadata = Base.metadata

# Generate migration from model diff
alembic revision --autogenerate -m "add articles table"

# Apply / rollback
alembic upgrade head
alembic downgrade -1

# Check current revision
alembic current
alembic history --verbose