sa_store.py en Python
Una capa de datos con SQLAlchemy: modelos, relaciones, consultas, agregados.
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 sembrar(sesion: Session) -> None:
basicos = Tour(slug="basics", title="Fundamentos del lenguaje")
basicos.steps = [
Step(title="Variables", target_tpm=80),
Step(title="Cadenas", target_tpm=85),
Step(title="Colecciones", target_tpm=90),
]
traits = Tour(slug="traits", title="Traits y genéricos")
traits.steps = [
Step(title="Objetos trait", target_tpm=110),
Step(title="Impls blanket", target_tpm=105),
]
sesion.add_all([basicos, traits])
sesion.commit()
def reportar(sesion: Session) -> None:
filas = sesion.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}{'pasos':>6}{'meta media':>13}")
for slug, steps, meta_media in filas:
print(f"{slug:<10}{steps:>6}{meta_media:>13.1f}")
dificil = sesion.scalars(
select(Step).order_by(Step.target_tpm.desc()).limit(1)).one()
print(f"paso más duro: {dificil.title} en {dificil.tour.slug}")
def principal():
engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as sesion:
sembrar(sesion)
reportar(sesion)
basicos = sesion.scalars(
select(Tour).where(Tour.slug == "basics")).one()
print(repr(basicos))
sesion.delete(basicos)
sesion.commit()
print("pasos restantes:",
sesion.scalar(select(func.count(Step.id))))
if __name__ == "__main__":
principal()
Cómo funciona
- Dos clases mapeadas con una relación en ambas direcciones.
- Una sesión por unidad de trabajo, confirmada una sola vez.
- El reporte es un select agrupado, no un bucle de consultas.
Palabras clave y builtins usados aquí
asclassdefforidifintlenlistpassprintreprreturnselfstrwith
El intento, en números
- Líneas
- 87
- Caracteres a escribir
- 2242
- Tokens
- 652
- Ritmo de tres estrellas
- 115 tpm
Al ritmo de tres estrellas de 115 tokens por minuto, este intento toma unos 340 segundos.
Paso 2 de 2 en Bis; paso 19 de 19 en Servicios web y acceso a datos.