SQLModel
SQLModel runs in the same process as the window (architecture 1 in Architectures); when the same models must be shared over HTTP, put them behind FastAPI as in the FastAPI demo. There are two demos:
sqlmodel_heroes: theTeamandHerotables of SQLModel’s tutorial, with its sample data,where()filters, relationships and a data model for the form.sqlmodel_inventory: an inventory whose validation rules and table come from one class.
The tutorial’s heroes in a window
Section titled “The tutorial’s heroes in a window”
cd examples/integrations/sqlmodel_heroesuv run main.pycd examples/integrations/sqlmodel_heroespython3 -m venv .venvsource .venv/bin/activatepip install pyzed-ui sqlmodelpython main.pycd examples/integrations/sqlmodel_heroespy -m venv .venv.venv\Scripts\Activate.ps1pip install pyzed-ui sqlmodelpython main.pycd examples/integrations/sqlmodel_heroespython3 -m pip install pyzed-ui sqlmodelpython3 main.pycd examples/integrations/sqlmodel_heroespy -m pip install pyzed-ui sqlmodelpy main.py-
Define the tables as in the tutorial’s Relationship Attributes chapter:
class Team(SQLModel, table=True):id: int | None = Field(default=None, primary_key=True)name: str = Field(index=True)headquarters: strheroes: 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: strage: 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") -
Create the engine and the tables, and add the tutorial’s heroes the first time:
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) -
Query with
select()andwhere(). Filters are added one by one;col()gives the type checker a column for the comparison, as the tutorial recommends: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))selectinloadloads each hero’s team before the session closes, so the table can showhero.team.name. -
Validate the form with a data model.
table=Truemodels don’t validate (SQLAlchemy creates them), so the tutorial’s Multiple Models approach applies: aHeroCreatewithouttable=Truechecks what the form typed.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 = Nonedef 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 Nonereturn Command.perform(lambda: db.save(hero_id, data), lambda _: self.reload()) -
Create or update with
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 pzfrom pyzed import Command, ui
import dbfrom 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)"""The tables of SQLModel's tutorial ("Relationship Attributes"): https://sqlmodel.tiangolo.com/tutorial/"""
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
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")
class HeroCreate(SQLModel): """A data model (no table=True) that validates what the form typed, as in the "Multiple Models" chapter."""
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
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)
def create_heroes(): """The tutorial's heroes and teams, added the first time.""" with Session(engine) as session: if session.exec(select(Hero)).first(): return team_preventers = Team(name="Preventers", headquarters="Sharp Tower") team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar") session.add_all([ Hero(name="Deadpond", secret_name="Dive Wilson", team=team_z_force), Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers), Hero(name="Spider-Boy", secret_name="Pedro Parqueador", team=team_preventers), Team(name="Wakaland", headquarters="Wakaland Capital City", heroes=[ Hero(name="Black Lion", secret_name="Trevor Challa", age=35), Hero(name="Princess Sure-E", secret_name="Sure-E"), ]), Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32, team=team_preventers), Hero(name="Dr. Weird", secret_name="Steve Weird", age=36, team=team_preventers), Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93, team=team_preventers), ]) session.commit()"""Queries the window runs (in worker threads): select(), where(), relationships, add, delete."""
from sqlalchemy.orm import selectinloadfrom sqlmodel import Session, col, select
from models import Hero, HeroCreate, Team, create_db_and_tables, create_heroes, engine
def setup() -> None: create_db_and_tables() create_heroes()
def teams() -> list[Team]: with Session(engine) as session: # Load each team's heroes now: relationships can't be read after the session closes. return list(session.exec(select(Team).options(selectinload(Team.heroes)).order_by(Team.name)))
def heroes(team_id: int | None | str, min_age: int | None) -> list[Hero]: """team_id is a team id, None for heroes without a team, or "all".""" 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) # the tutorial's "where" chapter with Session(engine) as session: return list(session.exec(statement))
def save(hero_id: int | None, data: HeroCreate) -> None: 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()
def delete(hero_id: int) -> None: with Session(engine) as session: session.delete(session.get(Hero, hero_id)) session.commit()One model for the form and the table: the inventory demo
Section titled “One model for the form and the table: the inventory demo”SQLModel joins Pydantic and SQLAlchemy: one class describes the fields, their rules and the table. In a desktop app that means the form and the database agree by construction. This demo is an inventory with stock levels, a donut chart and a dialog whose fields show Pydantic’s errors.

cd examples/integrations/sqlmodel_inventoryuv run main.pycd examples/integrations/sqlmodel_inventorypython3 -m venv .venvsource .venv/bin/activatepip install pyzed-ui sqlmodelpython main.pycd examples/integrations/sqlmodel_inventorypy -m venv .venv.venv\Scripts\Activate.ps1pip install pyzed-ui sqlmodelpython main.pycd examples/integrations/sqlmodel_inventorypython3 -m pip install pyzed-ui sqlmodelpython3 main.pycd examples/integrations/sqlmodel_inventorypy -m pip install pyzed-ui sqlmodelpy main.py-
Put the rules in a base class. The table and the form’s validator both inherit from it:
class ProductBase(SQLModel):name: str = Field(min_length=2, max_length=60)sku: strcategory: str = "Keyboards"price: float = Field(gt=0)stock: int = Field(ge=0)@field_validator("sku")@classmethoddef 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 skuclass 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.""" -
Validate with
ProductIn, notProduct. SQLModel skips validation ontable=Truemodels (SQLAlchemy builds them), so a separate class withouttable=Truedoes it. The form keeps everything as text and Pydantic coerces it:"12.5"becomes12.5,"abc"becomes an error.def save(self):try:data = ProductIn.model_validate(self.draft)except ValidationError as error:self.errors = pz.validation_errors(error)return Noneproduct_id = None if self.editing == "new" else self.editing.idreturn Command.perform(lambda: models.save(product_id, data), self.saved) -
Store it in a thread.
models.savecopies the validated data onto the row withsqlmodel_updateand returns{field: message}when something only the database knows is wrong, like a duplicated SKU: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 -
Show the errors in a dialog.
ui.layerputs aui.modalover the page; eachui.fieldtakes its message fromself.errors:ui.field("Price", ui.input(self.draft["price"], self.edit("price")), error=self.errors.get("price")) -
Render cells with widgets. A
ui.Columncan draw anything; the stock column shows buttons and a badge that turns amber or red when stock runs low: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 pzfrom pyzed import Command, ui
import modelsfrom 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)"""SQLModel models: one class is both the Pydantic schema and the database table."""
from __future__ import annotations
import refrom pathlib import Path
from pydantic import field_validatorfrom sqlmodel import Field, Session, SQLModel, create_engine, select
CATEGORIES = ["Keyboards", "Mice", "Monitors", "Audio", "Cables"]
engine = create_engine(f"sqlite:///{Path(__file__).with_name('inventory.db')}")
class ProductBase(SQLModel): """The fields and their rules, shared by the form and the table."""
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: str) -> str: 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): # Table models skip validation on purpose (SQLAlchemy builds them); validate with ProductIn first. id: int | None = Field(default=None, primary_key=True) sku: str = Field(unique=True, index=True)
class ProductIn(ProductBase): """Validates what the form typed: "12.5" becomes 12.5, "abc" becomes an error on that field."""
def setup() -> None: SQLModel.metadata.create_all(engine) with Session(engine) as session: if session.exec(select(Product)).first(): return session.add_all([ Product(name="Mechanical keyboard", sku="KEY-0001", category="Keyboards", price=129.0, stock=14), Product(name="Low-profile keyboard", sku="KEY-0002", category="Keyboards", price=89.5, stock=3), Product(name="Wireless mouse", sku="MOU-0001", category="Mice", price=49.9, stock=40), Product(name="Vertical mouse", sku="MOU-0002", category="Mice", price=59.0, stock=0), Product(name='27" 4K monitor', sku="MON-0001", category="Monitors", price=399.0, stock=6), Product(name='34" ultrawide', sku="MON-0002", category="Monitors", price=649.0, stock=2), Product(name="Studio headphones", sku="AUD-0001", category="Audio", price=149.0, stock=11), Product(name="USB microphone", sku="AUD-0002", category="Audio", price=99.0, stock=7), Product(name="USB-C cable 2 m", sku="CAB-0001", category="Cables", price=12.0, stock=120), Product(name="HDMI 2.1 cable", sku="CAB-0002", category="Cables", price=18.0, stock=4), ]) session.commit()
def products() -> list[Product]: with Session(engine) as session: return list(session.exec(select(Product).order_by(Product.category, Product.name)))
def save(product_id: int | None, data: ProductIn) -> Product | dict: """Stores validated data; returns ``{field: message}`` when the SKU is taken.""" 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
def restock(product_id: int, amount: int) -> None: with Session(engine) as session: product = session.get(Product, product_id) product.stock = max(0, product.stock + amount) session.add(product) session.commit()