Ir al contenido
Pyzedv1.9.0

SQLModel

SQLModel corre en el mismo proceso que la ventana (la arquitectura 1 de Arquitecturas); cuando los mismos modelos se tienen que compartir por HTTP, ponlos detrás de FastAPI como en la demo de FastAPI. Hay dos demos:

  • sqlmodel_heroes: las tablas Team y Hero del tutorial de SQLModel, con sus datos de ejemplo, filtros where(), relaciones y un modelo de datos para el formulario.
  • sqlmodel_inventory: un inventario cuyas reglas de validación y cuya tabla salen de una sola clase.

Los héroes del tutorial en una ventana

Sección titulada «Los héroes del tutorial en una ventana»

Héroes agrupados por equipo, filtrados por edad, con un panel de edición

Windows, macOS y Linux
cd examples/integrations/sqlmodel_heroes
uv run main.py

  1. Define las tablas como en el capítulo Relationship Attributes del tutorial:

    class Team(SQLModel, table=True):
        id: int | None = Field(default=None, primary_key=True)
        name: str = Field(index=True)
        headquarters: str
    

        heroes: list["Hero"] = Relationship(back_populates="team")
    

    class Hero(SQLModel, table=True):
        id: int | None = Field(default=None, primary_key=True)
        name: str = Field(index=True)
        secret_name: str
        age: int | None = Field(default=None, index=True)
    

        team_id: int | None = Field(default=None, foreign_key="team.id")
        team: Team | None = Relationship(back_populates="heroes")

  2. Crea el engine y las tablas, y añade los héroes del tutorial la primera vez:

    sqlite_file_name = "database.db"
    sqlite_url = f"sqlite:///{sqlite_file_name}"
    engine = create_engine(sqlite_url)
    

    def create_db_and_tables():
        SQLModel.metadata.create_all(engine)

  3. Consulta con select() y where(). Los filtros se añaden uno a uno; col() le da al comprobador de tipos una columna para la comparación, como recomienda el tutorial:

    def heroes(team_id, min_age):
        statement = select(Hero).options(selectinload(Hero.team)).order_by(Hero.name)
        if team_id != "all":
            statement = statement.where(Hero.team_id == team_id)
        if min_age is not None:
            statement = statement.where(col(Hero.age) >= min_age)
        with Session(engine) as session:
            return list(session.exec(statement))

    selectinload carga el equipo de cada héroe antes de que se cierre la sesión, para que la tabla pueda mostrar hero.team.name.

  4. Valida el formulario con un modelo de datos. Los modelos con table=True no validan (los crea SQLAlchemy), así que se aplica el enfoque de Multiple Models del tutorial: un HeroCreate sin table=True comprueba lo que se escribió en el formulario.

    class HeroCreate(SQLModel):
        name: str = Field(min_length=1)
        secret_name: str = Field(min_length=1)
        age: int | None = Field(default=None, ge=0)
        team_id: int | None = None
    

    def save(self):
        try:
            data = HeroCreate.model_validate({**self.draft, "age": self.draft["age"] or None, ...})
        except ValidationError as error:
            self.errors = pz.validation_errors(error)
            return None
        return Command.perform(lambda: db.save(hero_id, data), lambda _: self.reload())

  5. Crea o actualiza con sqlmodel_update:

    def save(hero_id, data: HeroCreate):
        with Session(engine) as session:
            hero = session.get(Hero, hero_id) if hero_id else Hero.model_validate(data)
            hero.sqlmodel_update(data.model_dump())
            session.add(hero)
            session.commit()
"""SQLModel's tutorial in a Pyzed window: teams, heroes, where() filters and a validated form.


    uv run main.py


The database (database.db) is created and filled with the tutorial's heroes on the first run.
Walkthrough: https://albakore.github.io/pyzed/integrations/sqlmodel/
"""


from __future__ import annotations


from pydantic import ValidationError


import pyzed as pz
from pyzed import Command, ui


import db
from models import Hero, HeroCreate


NO_TEAM = "No team"




class Heroes(pz.App):
    title = "Heroes · SQLModel"
    size = (1060, 620)


    def __init__(self):
        self.teams = []
        self.heroes: list[Hero] = []
        self.team_id: int | None | str = "all"  # a team id, None (heroes without a team) or "all"
        self.older = False                       # only heroes aged 35 or more
        self.editing: Hero | None | str = None   # a Hero, "new" or None
        self.draft: dict = {}
        self.errors: dict[str, str] = {}


    # -- database, in worker threads ------------------------------------------------------------
    def on_start(self):
        return Command.perform(db.setup, lambda _: self.reload())


    def reload(self):
        team_id, min_age = self.team_id, 35 if self.older else None
        return [Command.perform(db.teams, lambda teams: setattr(self, "teams", teams)),
                Command.perform(lambda: db.heroes(team_id, min_age), lambda heroes: setattr(self, "heroes", heroes))]


    def show_team(self, team_id):
        self.team_id = team_id
        return self.reload()


    def toggle_older(self, value: str):
        self.older = value != "All ages"
        return self.reload()


    # -- the form ---------------------------------------------------------------------------------
    def open(self, hero):
        self.editing, self.errors = hero, {}
        if hero == "new":
            self.draft = {"name": "", "secret_name": "", "age": "", "team": NO_TEAM}
        else:
            self.draft = {"name": hero.name, "secret_name": hero.secret_name,
                          "age": "" if hero.age is None else str(hero.age),
                          "team": hero.team.name if hero.team else NO_TEAM}


    def save(self):
        team = next((t for t in self.teams if t.name == self.draft["team"]), None)
        try:
            data = HeroCreate.model_validate({**self.draft, "age": self.draft["age"] or None,
                                              "team_id": team.id if team else None})
        except ValidationError as error:
            self.errors = pz.validation_errors(error)
            return None
        hero_id = None if self.editing == "new" else self.editing.id
        self.editing, self.errors = None, {}
        return Command.perform(lambda: db.save(hero_id, data), lambda _: self.reload())


    def delete(self):
        hero_id, self.editing = self.editing.id, None
        return Command.perform(lambda: db.delete(hero_id), lambda _: self.reload())


    def edit(self, key):
        return lambda value: self.draft.__setitem__(key, value)


    # -- view ---------------------------------------------------------------------------------------
    def sidebar(self, t):
        def item(label, team_id, count, icon):
            return ui.list_item(label, lambda: self.show_team(team_id), icon=icon,
                                end=ui.muted(str(count), size="sm"), selected=self.team_id == team_id, inset=True, height=30)


        teams = [item(team.name, team.id, len(team.heroes), "shield") for team in self.teams]
        return pz.container(pz.column([
            ui.muted("TEAMS", size="xs", weight="semibold"),
            item("All heroes", "all", sum(len(t.heroes) for t in self.teams), "users"),
            *teams,
            item(NO_TEAM, None, "", "user-x"),
        ], spacing=2), padding=8, width=230, height="fill", bg=t.panel)


    def panel(self, t):
        if self.editing is None:
            return None
        e = self.errors.get
        new = self.editing == "new"
        return pz.container(pz.column([
            ui.heading("New hero" if new else self.editing.name, level=3),
            ui.field("Name", ui.input(self.draft["name"], self.edit("name"), width=180), error=e("name")),
            ui.field("Secret name", ui.input(self.draft["secret_name"], self.edit("secret_name"), width=180),
                     error=e("secret_name")),
            ui.field("Age", ui.input(self.draft["age"], self.edit("age"), placeholder="unknown", width=180), error=e("age")),
            ui.field("Team", ui.select([NO_TEAM, *[team.name for team in self.teams]], self.draft["team"],
                                       self.edit("team"), width=180)),
            pz.row([None if new else ui.button("Delete", self.delete, icon="trash-2", variant="ghost", tint="error"),
                    pz.hspace(), ui.button("Cancel", lambda: setattr(self, "editing", None), variant="ghost"),
                    ui.button("Save", self.save, icon="check", variant="solid")], spacing=6),
        ], spacing=12), padding=16, width=380, height="fill", bg=t.panel)


    def view(self):
        t = self.resolved_theme()
        columns = [ui.Column("Name", "name", width="fill:2"), ui.Column("Secret name", "secret_name", width="fill:2"),
                   ui.Column("Age", lambda h: "—" if h.age is None else h.age, width=60, align="end"),
                   ui.Column("Team", lambda h: h.team.name if h.team else "—")]
        main = pz.column([
            pz.row([ui.segmented(["All ages", "Age ≥ 35"], "Age ≥ 35" if self.older else "All ages", self.toggle_older),
                    pz.hspace(), ui.button("New hero", lambda: self.open("new"), icon="plus", variant="solid")],
                   align="center"),
            ui.table(columns, self.heroes, on_row_press=self.open, selected=self.editing, height="fill"),
        ], spacing=12, padding=16)
        return pz.row([self.sidebar(t), pz.container(main, width="fill", height="fill", bg=t.editor), self.panel(t)])




if __name__ == "__main__":
    pz.run(Heroes)

Un modelo para el formulario y la tabla: la demo de inventario

Sección titulada «Un modelo para el formulario y la tabla: la demo de inventario»

SQLModel une Pydantic y SQLAlchemy: una sola clase describe los campos, sus reglas y la tabla. En una app de escritorio eso significa que el formulario y la base de datos coinciden por diseño. Esta demo es un inventario con niveles de stock, un gráfico de dona y un diálogo cuyos campos muestran los errores de Pydantic.

Una tabla de inventario con badges de stock y un diálogo de edición con errores de validación

Windows, macOS y Linux
cd examples/integrations/sqlmodel_inventory
uv run main.py

  1. Pon las reglas en una clase base. La tabla y el validador del formulario heredan de ella:

    class ProductBase(SQLModel):
        name: str = Field(min_length=2, max_length=60)
        sku: str
        category: str = "Keyboards"
        price: float = Field(gt=0)
        stock: int = Field(ge=0)
    

        @field_validator("sku")
        @classmethod
        def sku_format(cls, sku):
            sku = sku.strip().upper()
            if not re.fullmatch(r"[A-Z]{3}-\d{4}", sku):
                raise ValueError("Expected something like KEY-0001")
            return sku
    

    class Product(ProductBase, table=True):
        id: int | None = Field(default=None, primary_key=True)
        sku: str = Field(unique=True, index=True)
    

    class ProductIn(ProductBase):
        """Validates what the form typed."""

  2. Valida con ProductIn, no con Product. SQLModel no valida los modelos con table=True (los construye SQLAlchemy), así que lo hace una clase aparte sin table=True. El formulario guarda todo como texto y Pydantic lo convierte: "12.5" pasa a ser 12.5 y "abc", un error.

    def save(self):
        try:
            data = ProductIn.model_validate(self.draft)
        except ValidationError as error:
            self.errors = pz.validation_errors(error)
            return None
        product_id = None if self.editing == "new" else self.editing.id
        return Command.perform(lambda: models.save(product_id, data), self.saved)

  3. Guárdalo en un hilo. models.save copia los datos validados en la fila con sqlmodel_update y devuelve {campo: mensaje} cuando algo que solo sabe la base de datos está mal, como un SKU repetido:

    def save(product_id, data):
        with Session(engine, expire_on_commit=False) as session:
            clash = session.exec(select(Product).where(Product.sku == data.sku, Product.id != product_id)).first()
            if clash:
                return {"sku": f"Already used by “{clash.name}”"}
            product = session.get(Product, product_id) if product_id else Product.model_validate(data)
            product.sqlmodel_update(data.model_dump())
            session.add(product)
            session.commit()
            return product

  4. Muestra los errores en un diálogo. ui.layer pone un ui.modal sobre la página; cada ui.field toma su mensaje de self.errors:

    ui.field("Price", ui.input(self.draft["price"], self.edit("price")), error=self.errors.get("price"))

  5. Dibuja las celdas con widgets. Una ui.Column puede dibujar cualquier cosa; la columna de stock muestra botones y un badge que se vuelve ámbar o rojo cuando queda poco stock:

    ui.Column("Stock", lambda p: p, width=180, align="center", render=self.stock_cell, sortable=False)
"""An inventory app on SQLModel: the form is validated by the same model that defines the table.


    uv run main.py


The database (inventory.db) is created next to this file on the first run.
Walkthrough: https://albakore.github.io/pyzed/integrations/sqlmodel/
"""


from __future__ import annotations


from pydantic import ValidationError


import pyzed as pz
from pyzed import Command, ui


import models
from models import CATEGORIES, Product, ProductIn


LOW_STOCK = 5




class Inventory(pz.App):
    title = "Inventory · SQLModel"
    size = (1100, 680)


    def __init__(self):
        self.products: list[Product] = []
        self.category = "All"
        self.editing = None            # a Product, "new" or None
        self.draft: dict[str, str] = {}
        self.errors: dict[str, str] = {}


    def on_start(self):
        return Command.perform(lambda: (models.setup(), models.products())[1], self.loaded)


    def loaded(self, products: list[Product]):
        self.products = products


    def reload(self):
        return Command.perform(models.products, self.loaded)


    # -- the form ----------------------------------------------------------------------
    def open(self, product):
        self.editing, self.errors = product, {}
        if product == "new":
            self.draft = {"name": "", "sku": "", "category": CATEGORIES[0], "price": "", "stock": "0"}
        else:
            self.draft = {key: str(getattr(product, key)) for key in ("name", "sku", "category", "price", "stock")}


    def edit(self, key: str):
        return lambda value: self.draft.__setitem__(key, value)


    def save(self):
        try:
            data = ProductIn.model_validate(self.draft)  # strings from the inputs are coerced or rejected here
        except ValidationError as error:
            self.errors = pz.validation_errors(error)
            return None
        product_id = None if self.editing == "new" else self.editing.id
        return Command.perform(lambda: models.save(product_id, data), self.saved)


    def saved(self, result):
        if isinstance(result, dict):
            self.errors = result
            return None
        self.editing = None
        return self.reload()


    def restock(self, product: Product, amount: int):
        return Command.perform(lambda: models.restock(product.id, amount), lambda _: self.reload())


    def on_key(self, combo):
        if combo == "escape":
            self.editing = None
        elif combo == "ctrl-n":
            self.open("new")


    # -- view ----------------------------------------------------------------------------
    def stock_cell(self, product: Product):
        if product.stock == 0:
            badge = ui.badge("out of stock", tint="error")
        elif product.stock < LOW_STOCK:
            badge = ui.badge(f"{product.stock} left", tint="warning")
        else:
            badge = ui.label(str(product.stock), mono=True)
        return pz.row([
            ui.icon_button("minus", lambda: self.restock(product, -1), size="sm", tooltip="Sell one"),
            pz.container(badge, width=96, align_x="center"),
            ui.icon_button("plus", lambda: self.restock(product, 1), size="sm", tooltip="Receive one"),
        ], spacing=4, align="center")


    def summary(self, t):
        value = sum(p.price * p.stock for p in self.products)
        by_category = [sum(p.price * p.stock for p in self.products if p.category == c) for c in CATEGORIES]
        palette = [t.accent, t.success, t.warning, t.error, t.players[3]]
        legend = pz.column([pz.row([ui.indicator(color), ui.muted(c, size="sm")], spacing=6, align="center")
                            for c, color in zip(CATEGORIES, palette)], spacing=2)
        return pz.row([
            ui.stat("Stock value", f"${value:,.0f}"),
            ui.stat("Low stock", sum(0 < p.stock < LOW_STOCK for p in self.products)),
            ui.stat("Out of stock", sum(p.stock == 0 for p in self.products)),
            ui.card(pz.row([ui.donut(by_category, colors=palette, size=84, thickness=12), legend],
                           spacing=14, align="center"), padding=10),
        ], spacing=12, align="center")


    def dialog(self):
        if self.editing is None:
            return None
        e = self.errors.get
        fields = pz.column([
            ui.field("Name", ui.input(self.draft["name"], self.edit("name"), width=240), error=e("name")),
            ui.field("SKU", ui.input(self.draft["sku"], self.edit("sku"), placeholder="KEY-0003", mono=True,
                                     width=240), description="Three letters, a dash and four digits", error=e("sku")),
            ui.field("Category", ui.select(CATEGORIES, self.draft["category"], self.edit("category"), width=240)),
            ui.field("Price", ui.input(self.draft["price"], self.edit("price"), placeholder="0.00", width=240),
                     error=e("price")),
            ui.field("Stock", ui.input(self.draft["stock"], self.edit("stock"), width=240), error=e("stock")),
        ], spacing=12)
        close = lambda: setattr(self, "editing", None)  # noqa: E731
        return ui.modal(fields, title="New product" if self.editing == "new" else "Edit product", on_close=close,
                        width=520, footer=[ui.button("Cancel", close, variant="ghost"),
                                           ui.button("Save", self.save, icon="check", variant="solid")])


    def view(self):
        t = self.resolved_theme()
        shown = [p for p in self.products if self.category in ("All", p.category)]
        columns = [
            ui.Column("Product", "name", width="fill:3"),
            ui.Column("SKU", "sku", render=lambda sku: ui.label(sku, mono=True, color="text_muted")),
            ui.Column("Category", "category"),
            ui.Column("Price", "price", align="end", render=lambda price: ui.label(f"${price:,.2f}", mono=True)),
            ui.Column("Stock", lambda p: p, width=180, align="center", render=self.stock_cell, sortable=False),
        ]
        toolbar = pz.row([
            ui.segmented(["All", *CATEGORIES], self.category, lambda c: setattr(self, "category", c)),
            pz.hspace(),
            ui.button("New product", lambda: self.open("new"), icon="package-plus", variant="solid", keys="ctrl-n"),
        ], align="center")
        page = pz.container(
            pz.column([self.summary(t), toolbar,
                       ui.table(columns, shown, on_row_press=self.open, row_height=40, height="fill")], spacing=14),
            padding=16, width="fill", height="fill", bg=t.editor,
        )
        return ui.layer(page, self.dialog(), on_dismiss=lambda: setattr(self, "editing", None),
                        backdrop="#00000066")




if __name__ == "__main__":
    pz.run(Inventory)