typestar

sa_store.py in Python

A SQLAlchemy data layer: models, relationships, queries, aggregates.

from sqlalchemy import ForeignKey, create_engine, func, select
from sqlalchemy.orm import (DeclarativeBase, Mapped, Session, mapped_column,
                            relationship)


class Base(DeclarativeBase):
    pass


class Tour(Base):
    __tablename__ = "tours"

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(unique=True)
    title: Mapped[str]
    steps: Mapped[list["Step"]] = relationship(
        back_populates="tour", cascade="all, delete-orphan")

    def __repr__(self) -> str:
        return f"Tour({self.slug!r}, {len(self.steps)} steps)"


class Step(Base):
    __tablename__ = "steps"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    target_tpm: Mapped[int] = mapped_column(default=90)
    tour_id: Mapped[int] = mapped_column(ForeignKey("tours.id"))
    tour: Mapped[Tour] = relationship(back_populates="steps")


def seed(session: Session) -> None:
    basics = Tour(slug="basics", title="Language basics")
    basics.steps = [
        Step(title="Variables", target_tpm=80),
        Step(title="Strings", target_tpm=85),
        Step(title="Collections", target_tpm=90),
    ]
    traits = Tour(slug="traits", title="Traits & generics")
    traits.steps = [
        Step(title="Trait objects", target_tpm=110),
        Step(title="Blanket impls", target_tpm=105),
    ]
    session.add_all([basics, traits])
    session.commit()


def report(session: Session) -> None:
    rows = session.execute(
        select(Tour.slug,
               func.count(Step.id).label("steps"),
               func.avg(Step.target_tpm).label("mean_target"))
        .join(Tour.steps)
        .group_by(Tour.slug)
        .order_by(func.count(Step.id).desc())
    ).all()

    print(f"{'tour':<10}{'steps':>6}{'mean target':>13}")
    for slug, steps, mean_target in rows:
        print(f"{slug:<10}{steps:>6}{mean_target:>13.1f}")

    hardest = session.scalars(
        select(Step).order_by(Step.target_tpm.desc()).limit(1)).one()
    print(f"hardest step: {hardest.title} in {hardest.tour.slug}")


def main():
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        seed(session)
        report(session)

        basics = session.scalars(
            select(Tour).where(Tour.slug == "basics")).one()
        print(repr(basics))

        session.delete(basics)
        session.commit()
        print("steps left:",
              session.scalar(select(func.count(Step.id))))


if __name__ == "__main__":
    main()

How it works

  1. Two mapped classes with a relationship in both directions.
  2. A session per unit of work, committed once.
  3. The report is one grouped select, not a loop of queries.

Keywords and builtins used here

The run, in numbers

Lines
87
Characters to type
2214
Tokens
652
Three-star pace
115 tpm

At the three-star pace of 115 tokens a minute, this run takes about 340 seconds.

Type this snippet

Step 2 of 2 in Encore, step 19 of 19 in Web services & data access.

← Previous