# SQLAlchemy

> El ORM Quick Start de SQLAlchemy en una app de escritorio de Pyzed, con una Session síncrona en hilos de trabajo o una AsyncSession en el loop de asyncio de Pyzed, más una tabla de 20.000 filas.

Source: https://albakore.github.io/pyzed/es/integrations/sqlalchemy/

SQLAlchemy corre en el mismo proceso que la ventana (la arquitectura 1 de
[Arquitecturas](https://albakore.github.io/pyzed/es/integrations/architectures/)). Hay dos demos:

- **`sqlalchemy_quickstart`**: los modelos `User` y `Address` del
  [ORM Quick Start](https://docs.sqlalchemy.org/en/20/orm/quickstart.html) de SQLAlchemy, con la
  misma ventana sobre una `Session` síncrona o sobre la [extensión asyncio](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html).
- **`sqlalchemy_contacts`**: 20.000 filas en una tabla virtual, con búsqueda y orden en SQL.

## El Quick Start en una ventana

(Image: Usuarios y sus direcciones de email del Quick Start de SQLAlchemy)

1. **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:

```python
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")
```

2. **Crea un Engine y las tablas:**

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

3. **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:

```python
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()
```

4. **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:

```python
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:

```python
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))
```

**Note:**

Con `AsyncSession`, la carga perezosa necesitaría un `await`, así que las relaciones se cargan
de antemano con `selectinload`, y `expire_on_commit=False` mantiene los objetos legibles después
de `commit()`. Instala `sqlalchemy[asyncio]` y `aiosqlite`; el extra trae `greenlet`, que la
extensión necesita.

| | `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) |

**main.py:**
```py
"""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)
```

**models.py:**
```py
"""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"),
    ]
```

**sync_backend.py:**
```py
"""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()
```

**async_backend.py:**
```py
"""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.

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

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.

1. **Los modelos son SQLAlchemy 2.0 normal.**

```python
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="")
```

2. **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:

```python
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))
```

3. **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:

```python
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.

4. **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`:

```python
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")
```

5. **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:

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

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

**Tip:**

¿Prefieres `asyncio`? La `AsyncSession` de SQLAlchemy con `aiosqlite` o `asyncpg` también
funciona: devuelve la corrutina en lugar de usar `Command.perform`. Mira
[Otras librerías](https://albakore.github.io/pyzed/es/integrations/other-libraries/#sqlalchemy-asyncio).

**main.py:**
```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)
```

**db.py:**
```py
"""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()
```