SQLAlchemy

SQLAlchemy corre en el mismo proceso que la ventana (la arquitectura 1 de Arquitecturas). Hay dos demos:

sqlalchemy_quickstart: los modelos User y Address del ORM Quick Start de SQLAlchemy, con la misma ventana sobre una Session síncrona o sobre la extensión asyncio.
sqlalchemy_contacts: 20.000 filas en una tabla virtual, con búsqueda y orden en SQL.

El Quick Start en una ventana

Usuarios y sus direcciones de email del Quick Start de SQLAlchemy

cd examples/integrations/sqlalchemy_quickstart
uv run main.py            # Session en hilos de trabajo
uv run main.py --async    # AsyncSession con aiosqlite en el loop de Pyzed

cd examples/integrations/sqlalchemy_quickstart
python3 -m venv .venv
source .venv/bin/activate
pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
python main.py            # Session en hilos de trabajo
python main.py --async    # AsyncSession con aiosqlite en el loop de Pyzed

cd examples/integrations/sqlalchemy_quickstart
py -m venv .venv
.venv\Scripts\Activate.ps1
pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
python main.py            # Session en hilos de trabajo
python main.py --async    # AsyncSession con aiosqlite en el loop de Pyzed

cd examples/integrations/sqlalchemy_quickstart
python3 -m pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
python3 main.py            # Session en hilos de trabajo
python3 main.py --async    # AsyncSession con aiosqlite en el loop de Pyzed

cd examples/integrations/sqlalchemy_quickstart
py -m pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
py main.py            # Session en hilos de trabajo
py main.py --async    # AsyncSession con aiosqlite en el loop de Pyzed

Declara los modelos igual que el Quick Start, con DeclarativeBase, Mapped y relationship. La cascada borra las direcciones de un usuario junto con el usuario:

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user", cascade="all, delete-orphan")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")

Crea un Engine y las tablas:

engine = create_engine("sqlite:///quickstart.db")
Base.metadata.create_all(engine)

Abre una Session por unidad de trabajo. Cada función del backend abre su propia Session, así que se puede ejecutar sin riesgo en cualquier hilo de trabajo. selectinload carga las direcciones antes de que se cierre la sesión, para que la ventana pueda leer user.addresses después:

def users() -> list[User]:
    stmt = select(User).options(selectinload(User.addresses)).order_by(User.name)
    with Session(engine) as session:
        return list(session.scalars(stmt))

def add_address(user_id: int, email_address: str) -> None:
    with Session(engine) as session:
        user = session.get(User, user_id)
        user.addresses.append(Address(email_address=email_address))
        session.commit()

Llámalo desde la ventana con Command.perform. La función corre en un hilo y then recibe el resultado en el hilo de la UI:

def on_start(self):
    return Command.perform(backend.setup, lambda _: self.reload())

def reload(self):
    return Command.perform(backend.users, self.loaded)

def add_address(self, *_):
    user_id, email = self.selected_id, self.new_email.strip()
    return Command.perform(lambda: backend.add_address(user_id, email), lambda _: self.reload())

La versión asyncio

async_backend.py tiene las mismas funciones escritas con create_async_engine, un async_sessionmaker y el driver aiosqlite. Command.perform ve que son funciones de corrutina y las ejecuta en el loop de asyncio de Pyzed en lugar de en un hilo, así que main.py no cambia:

engine = create_async_engine("sqlite+aiosqlite:///quickstart.db")
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def setup() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

async def users() -> list[User]:
    stmt = select(User).options(selectinload(User.addresses)).order_by(User.name)
    async with async_session() as session:
        return list(await session.scalars(stmt))

	`Session`	`AsyncSession`
Dónde corre	Un hilo de trabajo por llamada	El loop de asyncio de Pyzed
Llamada de Pyzed	`Command.perform(función_síncrona, then)`	`Command.perform(función_async, then)` o devolver la corrutina
Úsala cuando	La mayoría de las apps de escritorio; cualquier driver	Ya usas drivers de asyncio (asyncpg, aiosqlite)

"""SQLAlchemy's ORM Quick Start in a Pyzed window: users and their email addresses.

    uv run main.py            # sync Session, run in worker threads (sync_backend.py)
    uv run main.py --async    # AsyncSession on Pyzed's asyncio loop (async_backend.py)

Both backends have the same functions and the window doesn't care which one it gets:
Command.perform runs a regular function in a thread and a coroutine on the loop.
Walkthrough: https://albakore.github.io/pyzed/integrations/sqlalchemy/
"""

from __future__ import annotations

import sys

import pyzed as pz
from pyzed import Command, ui

if "--async" in sys.argv:
    import async_backend as backend
else:
    import sync_backend as backend


class Quickstart(pz.App):
    title = "Users · SQLAlchemy"
    size = (900, 560)

    def __init__(self):
        self.users = []
        self.selected_id: int | None = None
        self.new_user = {"name": "", "fullname": ""}
        self.new_email = ""
        self.editing: tuple[int, str] | None = None  # (address id, text being typed)
        self.errors: dict[str, str] = {}

    # -- database ----------------------------------------------------------------------------
    def on_start(self):
        return Command.perform(backend.setup, lambda _: self.reload())

    def reload(self):
        return Command.perform(backend.users, self.loaded)

    def loaded(self, users):
        self.users = users
        if self.selected_id not in {u.id for u in users}:
            self.selected_id = users[0].id if users else None

    def run(self, work):
        """Runs a backend call, then reloads. ``work`` is a lambda, so its arguments are read now."""
        return Command.perform(work, lambda _: self.reload())

    # -- actions -------------------------------------------------------------------------------
    def add_user(self, *_):
        name, fullname = self.new_user["name"].strip(), self.new_user["fullname"].strip()
        self.errors = {"name": "Required"} if not name else {"name": "30 characters at most"} if len(name) > 30 else {}
        if self.errors:
            return None
        self.new_user = {"name": "", "fullname": ""}
        return self.run(lambda: backend.add_user(name, fullname))

    def add_address(self, *_):
        email, self.new_email = self.new_email.strip(), ""
        user_id = self.selected_id
        return self.run(lambda: backend.add_address(user_id, email)) if "@" in email else None

    def save_address(self, *_):
        address_id, email = self.editing
        self.editing = None
        return self.run(lambda: backend.rename_address(address_id, email.strip()))

    def delete_user(self):
        user_id = self.selected_id
        return self.run(lambda: backend.delete_user(user_id))

    # -- view ----------------------------------------------------------------------------------
    def sidebar(self, t):
        rows = [ui.list_item(u.name, lambda u=u: setattr(self, "selected_id", u.id), icon="user",
                             end=ui.count_badge(len(u.addresses), tint="text_muted") if u.addresses else None,
                             selected=u.id == self.selected_id, inset=True, height=30) for u in self.users]
        form = pz.column([
            ui.input(self.new_user["name"], lambda v: self.new_user.__setitem__("name", v), placeholder="name",
                     on_submit=self.add_user),
            ui.label(self.errors["name"], size="sm", color="error") if self.errors.get("name") else None,
            ui.input(self.new_user["fullname"], lambda v: self.new_user.__setitem__("fullname", v),
                     placeholder="Full name (optional)", on_submit=self.add_user),
            ui.button("Add user", self.add_user, icon="user-plus", full_width=True),
        ], spacing=6)
        return pz.container(pz.column([ui.muted("user_account", size="xs", mono=True), *rows, pz.vspace(), form],
                                      spacing=2), padding=8, width=260, height="fill", bg=t.panel)

    def address_row(self, address):
        if self.editing and self.editing[0] == address.id:
            return pz.row([
                ui.input(self.editing[1], lambda v: setattr(self, "editing", (address.id, v)), on_submit=self.save_address),
                ui.icon_button("check", self.save_address, tooltip="Save"),
            ], spacing=6, align="center")
        return ui.list_item(address.email_address, lambda: setattr(self, "editing", (address.id, address.email_address)),
                            icon="mail", end=ui.muted("click to edit", size="sm"), inset=True, height=30)

    def detail(self):
        user = next((u for u in self.users if u.id == self.selected_id), None)
        if user is None:
            return ui.empty_state("No users", "Add one on the left.", icon="users")
        return pz.column([
            pz.row([ui.heading(user.fullname or user.name, level=2), pz.hspace(),
                    ui.button("Delete user", self.delete_user, icon="trash-2", variant="ghost", tint="error")],
                   align="center"),
            ui.muted(repr(user), mono=True, size="sm"),
            ui.label("Addresses", weight="semibold"),
            *[self.address_row(a) for a in user.addresses],
            pz.row([ui.input(self.new_email, lambda v: setattr(self, "new_email", v), placeholder="new@sqlalchemy.org",
                             icon="mail-plus", on_submit=self.add_address),
                    ui.button("Add", self.add_address)], spacing=6),
        ], spacing=10, padding=24)

    def view(self):
        t = self.resolved_theme()
        mode = "AsyncSession · aiosqlite" if backend.__name__ == "async_backend" else "Session · worker threads"
        return pz.column([
            pz.row([self.sidebar(t), pz.container(self.detail(), width="fill", height="fill", bg=t.editor)]),
            ui.status_bar([ui.status_item(mode, icon="database", icon_color="success")]),
        ])


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

"""The models of SQLAlchemy's ORM Quick Start: https://docs.sqlalchemy.org/en/20/orm/quickstart.html"""

from typing import List, Optional

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


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]

    addresses: Mapped[List["Address"]] = relationship(back_populates="user", cascade="all, delete-orphan")

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))

    user: Mapped["User"] = relationship(back_populates="addresses")

    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"


def sample_users() -> list[User]:
    """The users the Quick Start creates."""
    return [
        User(name="spongebob", fullname="Spongebob Squarepants",
             addresses=[Address(email_address="spongebob@sqlalchemy.org")]),
        User(name="sandy", fullname="Sandy Cheeks",
             addresses=[Address(email_address="sandy@sqlalchemy.org"), Address(email_address="sandy@squirrelpower.org")]),
        User(name="patrick", fullname="Patrick Star"),
    ]

"""The database code, as the Quick Start writes it: an Engine, a Session per unit of work, select().

Each function opens its own Session, so Pyzed can run it in a worker thread (Command.perform).
"""

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, selectinload

from models import Address, Base, User, sample_users

engine = create_engine("sqlite:///quickstart.db")


def setup() -> None:
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        if session.scalars(select(User)).first() is None:
            session.add_all(sample_users())
            session.commit()


def users() -> list[User]:
    # selectinload fetches the addresses now, so the window can read them after the session closes.
    stmt = select(User).options(selectinload(User.addresses)).order_by(User.name)
    with Session(engine) as session:
        return list(session.scalars(stmt))


def add_user(name: str, fullname: str) -> None:
    with Session(engine) as session:
        session.add(User(name=name, fullname=fullname or None))
        session.commit()


def add_address(user_id: int, email_address: str) -> None:
    with Session(engine) as session:
        user = session.get(User, user_id)
        user.addresses.append(Address(email_address=email_address))
        session.commit()


def rename_address(address_id: int, email_address: str) -> None:
    with Session(engine) as session:
        session.get(Address, address_id).email_address = email_address
        session.commit()


def delete_user(user_id: int) -> None:
    with Session(engine) as session:
        session.delete(session.get(User, user_id))  # cascade="all, delete-orphan" removes the addresses
        session.commit()

"""The same database code with SQLAlchemy's asyncio extension and the aiosqlite driver.

https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
The functions are coroutines: Pyzed runs them on its asyncio loop, where the engine lives.
"""

from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import selectinload

from models import Address, Base, User, sample_users

engine = create_async_engine("sqlite+aiosqlite:///quickstart.db")
async_session = async_sessionmaker(engine, expire_on_commit=False)


async def setup() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with async_session() as session:
        if (await session.scalars(select(User))).first() is None:
            session.add_all(sample_users())
            await session.commit()


async def users() -> list[User]:
    # Lazy loading needs an await, so load relationships eagerly with selectinload.
    stmt = select(User).options(selectinload(User.addresses)).order_by(User.name)
    async with async_session() as session:
        return list(await session.scalars(stmt))


async def add_user(name: str, fullname: str) -> None:
    async with async_session() as session:
        session.add(User(name=name, fullname=fullname or None))
        await session.commit()


async def add_address(user_id: int, email_address: str) -> None:
    async with async_session() as session:
        session.add(Address(user_id=user_id, email_address=email_address))
        await session.commit()


async def rename_address(address_id: int, email_address: str) -> None:
    async with async_session() as session:
        address = await session.get(Address, address_id)
        address.email_address = email_address
        await session.commit()


async def delete_user(user_id: int) -> None:
    async with async_session() as session:
        user = await session.get(User, user_id, options=[selectinload(User.addresses)])
        await session.delete(user)
        await session.commit()

Muchas filas: la demo de contactos

Un gestor de contactos sobre SQLAlchemy 2.0 y SQLite. La tabla tiene 20.000 filas y se desplaza con fluidez porque ui.table solo construye las filas visibles; la búsqueda y el orden se hacen en SQL, en un hilo de trabajo, así que la ventana nunca espera a la base de datos.

Una tabla de contactos con 20.000 filas, estadísticas y un panel de edición

cd examples/integrations/sqlalchemy_contacts
uv run main.py

cd examples/integrations/sqlalchemy_contacts
python3 -m venv .venv
source .venv/bin/activate
pip install pyzed-ui sqlalchemy
python main.py

cd examples/integrations/sqlalchemy_contacts
py -m venv .venv
.venv\Scripts\Activate.ps1
pip install pyzed-ui sqlalchemy
python main.py

cd examples/integrations/sqlalchemy_contacts
python3 -m pip install pyzed-ui sqlalchemy
python3 main.py

cd examples/integrations/sqlalchemy_contacts
py -m pip install pyzed-ui sqlalchemy
py main.py

La demo guarda el código de la base de datos en db.py, sin nada de Pyzed, y la ventana en main.py. Vale la pena copiar esa separación: el mismo db.py podría servir a una app de FastAPI o a una CLI.

Los modelos son SQLAlchemy 2.0 normal.

class Contact(Base):
    __tablename__ = "contacts"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(80), index=True)
    email: Mapped[str] = mapped_column(String(120), unique=True)
    company: Mapped[str] = mapped_column(String(80), default="")
    city: Mapped[str] = mapped_column(String(60), default="")

Cada consulta es una función con su propia sesión. Las sesiones de SQLAlchemy no se comparten entre hilos, así que cada función abre una y la cierra. Los objetos cargados conservan sus valores después de cerrar la sesión, que es todo lo que necesita la vista:

def search(text, sort, descending):
    column = COLUMNS[sort]
    query = select(Contact).order_by(column.desc() if descending else column, Contact.id)
    if text:
        pattern = f"%{text}%"
        query = query.where(or_(Contact.name.ilike(pattern), Contact.email.ilike(pattern),
                                Contact.company.ilike(pattern)))
    with Session(engine) as session:
        return list(session.scalars(query))

Ejecútala con Command.perform. La función corre en un hilo y then recibe el resultado en el hilo de la UI. Devolver una lista de comandos los ejecuta juntos:

def reload(self):
    query, sort, descending = self.query, self.sort, self.descending
    return [
        Command.perform(lambda: db.search(query, sort, descending), lambda rows: self.loaded(query, rows)),
        Command.perform(db.stats, lambda stats: setattr(self, "stats", stats)),
    ]

def loaded(self, query, rows):
    if query == self.query:          # a newer search is on its way: keep waiting for it
        self.rows, self.loading = rows, False

Copia primero en variables locales el estado que necesitas (query, sort, ...): la lambda se ejecuta más tarde, en otro hilo, y no debería leer self mientras el usuario sigue escribiendo.

Pasa todas las filas a ui.table. Está virtualizada, así que 20.000 filas cuestan lo mismo que 20. Al pulsar una cabecera se llama a on_sort, que aquí vuelve a ejecutar la consulta con un nuevo ORDER BY:
```
ui.table(columns, self.rows, sort_by=self.sort, descending=self.descending, on_sort=self.sort_by,
         on_row_press=self.open, selected=self.editing, striped=True, height="fill")
```

Convierte los errores de la base de datos en errores de campo. save devuelve el contacto o {campo: mensaje}. Una restricción de unicidad se convierte en un mensaje bajo el campo Email:

try:
    session.commit()
except IntegrityError:
    return {"email": "Another contact already uses this email"}

ui.field("Email", ui.input(self.draft["email"], ...), error=self.errors.get("email"))

main.py
db.py

"""A contacts manager on SQLAlchemy 2.0: 20,000 rows in a virtual table, searched and sorted in SQL.

    uv run main.py

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

from __future__ import annotations

import pyzed as pz
from pyzed import Command, ui

import db

FIELDS = [("name", "Name", "Ada Lovelace"), ("email", "Email", "ada@example.com"),
          ("company", "Company", "Analytical Engines"), ("city", "City", "London")]


class Contacts(pz.App):
    title = "Contacts · SQLAlchemy"
    size = (1180, 720)

    def __init__(self):
        self.rows: list[db.Contact] = []
        self.stats = {"total": 0, "companies": 0, "cities": []}
        self.query = ""
        self.sort, self.descending = "Name", False
        self.loading = True
        self.editing = None            # a Contact, "new" or None
        self.draft: dict[str, str] = {}
        self.errors: dict[str, str] = {}

    # -- database work runs in a thread with Command.perform ------------------------------
    def on_start(self):
        return Command.perform(db.setup, lambda _: self.reload())

    def reload(self):
        self.loading = True
        query, sort, descending = self.query, self.sort, self.descending
        return [
            Command.perform(lambda: db.search(query, sort, descending), lambda rows: self.loaded(query, rows)),
            Command.perform(db.stats, lambda stats: setattr(self, "stats", stats)),
        ]

    def loaded(self, query: str, rows: list[db.Contact]):
        if query == self.query:  # a newer search is on its way: keep waiting for it
            self.rows, self.loading = rows, False

    def search(self, text: str):
        self.query = text
        return self.reload()

    def sort_by(self, title: str):
        self.descending = not self.descending if title == self.sort else False
        self.sort = title
        return self.reload()

    # -- the edit panel ------------------------------------------------------------------
    def open(self, contact):
        self.editing, self.errors = contact, {}
        self.draft = {key: "" if contact == "new" else getattr(contact, key) for key, _, _ in FIELDS}

    def save(self):
        contact_id = None if self.editing == "new" else self.editing.id
        values = dict(self.draft)
        return Command.perform(lambda: db.save(contact_id, values), self.saved)

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

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

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

    # -- view ----------------------------------------------------------------------------
    def header(self):
        cities = self.stats["cities"]
        return pz.row([
            ui.stat("Contacts", f"{self.stats['total']:,}"),
            ui.stat("Companies", self.stats["companies"]),
            ui.card(pz.column([ui.muted("Top cities", size="sm"),
                               ui.bar_chart([count for _, count in cities], labels=[c.split()[0] for c, _ in cities],
                                            width=360, height=90)], spacing=6)),
        ], spacing=12)

    def panel(self, t):
        if self.editing is None:
            return None
        rows = [ui.field(title, ui.input(self.draft[key], lambda v, k=key: self.draft.__setitem__(k, v),
                                         placeholder=hint, width=220),
                         error=self.errors.get(key))
                for key, title, hint in FIELDS]
        new = self.editing == "new"
        return pz.container(
            pz.column([
                pz.row([ui.heading("New contact" if new else "Edit contact", level=3), pz.hspace(),
                        ui.icon_button("x", lambda: setattr(self, "editing", None), tooltip="Close (Esc)")],
                       align="center"),
                *rows,
                pz.row([None if new else ui.button("Delete", self.delete, icon="trash-2", variant="ghost", tint="error"),
                        pz.hspace(), ui.button("Save", self.save, icon="check", variant="solid")], spacing=8),
            ], spacing=14),
            padding=16, width=420, height="fill", bg=t.panel,
        )

    def view(self):
        t = self.resolved_theme()
        columns = [
            ui.Column("Name", "name", width="fill:3"),
            ui.Column("Email", "email", width="fill:3"),
            ui.Column("Company", "company", width="fill:2"),
            ui.Column("City", "city", width="fill:2"),
        ]
        toolbar = pz.row([
            ui.search_input(self.query, self.search, placeholder="Search by name, email or company…", width=360),
            ui.spinner() if self.loading else ui.muted(f"{len(self.rows):,} found", size="sm"),
            pz.hspace(),
            ui.button("New contact", lambda: self.open("new"), icon="user-plus", keys="ctrl-n"),
        ], spacing=10, align="center")
        # Rows come sorted from SQL; sort_by only draws the arrow (re-sorting a sorted list is cheap).
        table = ui.table(columns, self.rows, sort_by=self.sort, descending=self.descending, on_sort=self.sort_by,
                         on_row_press=self.open, selected=self.editing, striped=True, height="fill")
        main = pz.column([self.header(), toolbar, table], spacing=14, padding=16)
        return pz.container(pz.row([main, self.panel(t)]), width="fill", height="fill", bg=t.editor)


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

"""The database: SQLAlchemy 2.0 models on SQLite, plus the queries the window runs."""

from __future__ import annotations

import random
from datetime import datetime
from pathlib import Path

from sqlalchemy import String, create_engine, func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

DATABASE = Path(__file__).with_name("contacts.db")
engine = create_engine(f"sqlite:///{DATABASE}")


class Base(DeclarativeBase):
    pass


class Contact(Base):
    __tablename__ = "contacts"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(80), index=True)
    email: Mapped[str] = mapped_column(String(120), unique=True)
    company: Mapped[str] = mapped_column(String(80), default="")
    city: Mapped[str] = mapped_column(String(60), default="")
    created: Mapped[datetime] = mapped_column(default=datetime.now)


# Sorting by a column title of the table is an ORDER BY on this column.
COLUMNS = {"Name": Contact.name, "Email": Contact.email, "Company": Contact.company, "City": Contact.city}


def setup(rows: int = 20_000) -> None:
    """Creates the table and fills it with made-up contacts the first time."""
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        if session.scalar(select(func.count()).select_from(Contact)):
            return
        rng = random.Random(7)
        first = ["Ada", "Alan", "Grace", "Linus", "Guido", "Margaret", "Ken", "Barbara", "Dennis", "Radia",
                 "Tim", "Frances", "Bjarne", "Anita", "John", "Hedy", "Yukihiro", "Sophie", "Brendan", "Katherine"]
        last = ["Lovelace", "Turing", "Hopper", "Torvalds", "van Rossum", "Hamilton", "Thompson", "Liskov",
                "Ritchie", "Perlman", "Berners-Lee", "Allen", "Stroustrup", "Borg", "McCarthy", "Lamarr",
                "Matsumoto", "Wilson", "Eich", "Johnson"]
        companies = ["Acme", "Globex", "Initech", "Umbrella", "Hooli", "Stark", "Wayne", "Tyrell", "Cyberdyne", "Aperture"]
        cities = ["Buenos Aires", "Madrid", "Mexico City", "Bogotá", "Lima", "Santiago", "Montevideo", "Barcelona",
                  "Berlin", "Tokyo", "Toronto", "Lisbon"]
        weights = [12, 10, 9, 7, 6, 5, 4, 4, 3, 3, 2, 2]
        session.add_all(
            Contact(name=f"{rng.choice(first)} {rng.choice(last)}", email=f"user{i}@example.com",
                    company=rng.choice(companies), city=rng.choices(cities, weights)[0])
            for i in range(rows)
        )
        session.commit()


def search(text: str, sort: str, descending: bool) -> list[Contact]:
    """Contacts whose name, email or company contains ``text``, sorted in SQL."""
    column = COLUMNS[sort]
    query = select(Contact).order_by(column.desc() if descending else column, Contact.id)
    if text:
        pattern = f"%{text}%"
        query = query.where(or_(Contact.name.ilike(pattern), Contact.email.ilike(pattern),
                                Contact.company.ilike(pattern)))
    with Session(engine) as session:
        return list(session.scalars(query))  # plain objects keep their loaded values after the session closes


def stats() -> dict:
    """Totals and the five cities with more contacts, for the header."""
    with Session(engine) as session:
        total = session.scalar(select(func.count()).select_from(Contact))
        companies = session.scalar(select(func.count(func.distinct(Contact.company))))
        cities = session.execute(select(Contact.city, func.count()).group_by(Contact.city)
                                 .order_by(func.count().desc()).limit(5)).all()
    return {"total": total, "companies": companies, "cities": [(city, count) for city, count in cities]}


def save(contact_id: int | None, values: dict) -> Contact | dict:
    """Inserts or updates a contact. Returns it, or ``{field: message}`` when it can't be saved."""
    errors = {}
    if not values["name"].strip():
        errors["name"] = "A contact needs a name"
    if "@" not in values["email"]:
        errors["email"] = "Not an email address"
    if errors:
        return errors
    with Session(engine, expire_on_commit=False) as session:
        contact = session.get(Contact, contact_id) if contact_id else Contact()
        for key, value in values.items():
            setattr(contact, key, value.strip())
        session.add(contact)
        try:
            session.commit()
        except IntegrityError:
            return {"email": "Another contact already uses this email"}
        return contact


def delete(contact_id: int) -> None:
    with Session(engine) as session:
        contact = session.get(Contact, contact_id)
        if contact is not None:
            session.delete(contact)
            session.commit()