Database Modeling

Database Modeling Skill

> Expertise: PostgreSQL schema design, SQLAlchemy (async), query optimization, indexing, migrations (Alembic), safe schema changes.

Schema Design Patterns

Standard column set (all tables)

from sqlalchemy import Column, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False
    )

class Order(TimestampMixin, Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, index=True)
    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
    total_amount: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)

Soft delete pattern

class SoftDeleteMixin:
    deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

# Always filter in repository, never expose deleted records by default
class OrderRepository:
    async def list_active(self, session: AsyncSession):
        return await session.execute(
            select(Order).where(Order.deleted_at.is_(None))
        )

Indexing Strategy

-- Single column: high-cardinality columns used in WHERE/JOIN/ORDER BY
CREATE INDEX idx_orders_user_id ON orders(user_id);
CREATE INDEX idx_orders_status ON orders(status) WHERE deleted_at IS NULL;  -- partial index

-- Composite: query uses both columns together (order matters: equality first, then range)
CREATE INDEX idx_orders_user_created ON orders(user_id, created_at DESC);

-- Full-text search
CREATE INDEX idx_products_search ON products USING gin(to_tsvector('english', name || ' ' || description));

-- Never index: low-cardinality boolean columns, small tables (10k rows
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT * FROM orders WHERE user_id = 123 ORDER BY created_at DESC LIMIT 20;

-- Watch for: Seq Scan on large table → add index
--            Index Scan with high actual rows >> estimated rows → ANALYZE the table
--            Nested Loop with large inner side → consider Hash Join

Repository Pattern

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update

class OrderRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_by_id(self, order_id: int) -> Optional[Order]:
        result = await self.session.execute(
            select(Order).where(Order.id == order_id, Order.deleted_at.is_(None))
        )
        return result.scalar_one_or_none()

    async def list_by_user(
        self, user_id: int, *, limit: int = 20, cursor_id: Optional[int] = None
    ) -> list[Order]:
        q = select(Order).where(Order.user_id == user_id, Order.deleted_at.is_(None))
        if cursor_id:
            q = q.where(Order.id  None:
        await self.session.execute(
            update(Order).where(Order.id == order_id).values(status=status)
        )
        # No commit here — caller (service layer) owns the transaction

Migration Safety (Alembic)

# Generate migration
alembic revision --autogenerate -m "add_index_orders_user_id"

# ALWAYS review generated file before applying
alembic show head

# Apply
alembic upgrade head

# Rollback one step
alembic downgrade -1

Safe vs. unsafe schema operations

| Operation | Safe to deploy | Strategy | |---|---|---| | Add nullable column | ✅ Non-breaking | Apply directly | | Add column with default | ✅ (PostgreSQL 11+) | Apply directly | | Add NOT NULL column | ⚠️ Breaking | Add nullable → backfill → add constraint | | Add index | ✅ with CONCURRENTLY | CREATE INDEX CONCURRENTLY | | Rename column | ❌ Breaking | Expand/contract (add new → migrate code → drop old) | | Drop column | ❌ Breaking | Deprecate in code → drop in next release | | Change type | ❌ Breaking | Add new column with new type → migrate → drop old |

# Alembic: create index without locking table
def upgrade():
    op.execute("CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_orders_user_id ON orders(user_id)")

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_orders_user_id")

N+1 Query Prevention

# ❌ N+1: loads orders, then 1 query per order to get user
orders = await session.execute(select(Order)).scalars()
for order in orders:
    print(order.user.name)  # each access fires a query

# ✅ Eager load with joinedload
from sqlalchemy.orm import joinedload

orders = await session.execute(
    select(Order)
    .options(joinedload(Order.user))  # single JOIN
    .where(Order.status == "pending")
)

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: sawrus
• Source: sawrus/agent-guides
• License: MIT
• Homepage: https://sawrus.github.io/agent-guides

Install and usage instructions live in the source repository linked above.