ORMs and migrations

mkdir brew-orm
cd brew-orm
python -m venv .venv
# Windows PowerShell: .venv\Scripts\Activate.ps1
# macOS / Linux: source .venv/bin/activate
pip install sqlalchemy alembic

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite file for now. For Postgres use e.g.
# "postgresql+psycopg://postgres:devpass@localhost:5432/brewbean"
DATABASE_URL = "sqlite:///brewbean.db"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

from datetime import datetime, timezone

from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


class MenuItem(Base):
    __tablename__ = "menu_items"
    __table_args__ = (CheckConstraint("price > 0", name="ck_menu_items_price_positive"),)

    id: Mapped[str] = mapped_column(String, primary_key=True)
    name: Mapped[str] = mapped_column(String, nullable=False)
    category: Mapped[str] = mapped_column(String, nullable=False)
    price: Mapped[int] = mapped_column(Integer, nullable=False)


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_name: Mapped[str] = mapped_column(String, nullable=False)
    phone: Mapped[str] = mapped_column(String, nullable=False)
    status: Mapped[str] = mapped_column(String, nullable=False, default="received")
    pickup_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

    lines: Mapped[list["OrderLine"]] = relationship(
        back_populates="order", cascade="all, delete-orphan"
    )

    @property
    def total(self) -> int:
        return sum(line.qty * line.unit_price for line in self.lines)


class OrderLine(Base):
    __tablename__ = "order_lines"

    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id", ondelete="CASCADE"))
    item_id: Mapped[str] = mapped_column(ForeignKey("menu_items.id"))
    qty: Mapped[int] = mapped_column(Integer, nullable=False)
    unit_price: Mapped[int] = mapped_column(Integer, nullable=False)

    order: Mapped["Order"] = relationship(back_populates="lines")
    item: Mapped["MenuItem"] = relationship()

alembic init migrations

# in migrations/env.py, replace the line 'target_metadata = None' with:
from database import DATABASE_URL
from models import Base

config.set_main_option("sqlalchemy.url", DATABASE_URL)
target_metadata = Base.metadata

alembic revision --autogenerate -m "create menu, orders and order lines"
alembic upgrade head

    pickup_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

alembic revision --autogenerate -m "add pickup_time to orders"
alembic upgrade head
alembic current

alembic downgrade -1
alembic upgrade head

from database import SessionLocal
from models import MenuItem

MENU = [
    ("espresso", "Espresso", "coffee", 120),
    ("cappuccino", "Cappuccino", "coffee", 180),
    ("cold-brew", "Cold Brew", "coffee", 220),
    ("filter-coffee", "Filter Coffee", "coffee", 80),
    ("masala-chai", "Masala Chai", "tea", 90),
    ("croissant", "Butter Croissant", "bakery", 150),
    ("banana-bread", "Banana Bread", "bakery", 130),
    ("veg-sandwich", "Veg Grilled Sandwich", "food", 160),
]

with SessionLocal() as session:
    if session.query(MenuItem).count() == 0:
        session.add_all(
            MenuItem(id=i, name=n, category=c, price=p) for i, n, c, p in MENU
        )
        session.commit()
        print("Seeded", len(MENU), "menu items")
    else:
        print("Menu already seeded")

from sqlalchemy import func, select
from sqlalchemy.orm import selectinload

from database import SessionLocal
from models import MenuItem, Order, OrderLine

with SessionLocal() as session:
    # Create: an order with two lines, saved in one transaction
    prices = {m.id: m.price for m in session.scalars(select(MenuItem))}
    order = Order(customer_name="Asha", phone="9876543210")
    for item_id, qty in [("cappuccino", 1), ("croissant", 2)]:
        order.lines.append(OrderLine(item_id=item_id, qty=qty, unit_price=prices[item_id]))
    session.add(order)
    session.commit()
    print("Created order", order.id, "total", order.total)

    # Read: filter, sort
    coffees = session.scalars(
        select(MenuItem).where(MenuItem.category == "coffee").order_by(MenuItem.price.desc())
    ).all()
    print([c.name for c in coffees])

    # Read one by primary key, with its lines loaded in one extra query
    found = session.scalars(
        select(Order).where(Order.id == order.id).options(selectinload(Order.lines))
    ).one()
    print(found.customer_name, [(l.item_id, l.qty) for l in found.lines])

    # Update
    found.status = "preparing"
    session.commit()

    # Aggregate: units sold per category
    rows = session.execute(
        select(MenuItem.category, func.sum(OrderLine.qty))
        .join(OrderLine, OrderLine.item_id == MenuItem.id)
        .group_by(MenuItem.category)
    ).all()
    print(rows)

    # Delete
    session.delete(found)
    session.commit()
    print("Orders left:", session.scalar(select(func.count()).select_from(Order)))

python seed.py
python demo.py

from sqlalchemy import select
from sqlalchemy.orm import selectinload

stmt = select(Order).options(selectinload(Order.lines)).limit(50)

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session

from database import SessionLocal
from models import MenuItem

app = FastAPI()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/menu")
def get_menu(category: str | None = None, db: Session = Depends(get_db)):
    query = select(MenuItem).order_by(MenuItem.name)
    if category:
        query = query.where(MenuItem.category == category)
    items = db.scalars(query).all()
    return [
        {"id": m.id, "name": m.name, "category": m.category, "price": m.price}
        for m in items
    ]

pip install "psycopg[binary]"

DATABASE_URL = "postgresql+psycopg://postgres:devpass@localhost:5432/brewbean"

You are a senior backend engineer. Review my SQLAlchemy 2 models and the Alembic migration generated from them.

Models:
---
[paste models.py]
---

Migration:
---
[paste the migration file]
---

Tell me: (1) does the migration match the models, (2) is anything dangerous for a table that already has data, (3) missing indexes or constraints, (4) possible N+1 queries. Give corrected code for each problem.