# SQLAlchemy

> SQLAlchemy's ORM Quick Start in a Pyzed desktop app, with a sync Session in worker threads or an AsyncSession on Pyzed's asyncio loop, plus a 20,000-row table.

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

SQLAlchemy runs in the same process as the window (architecture 1 in
[Architectures](https://albakore.github.io/pyzed/integrations/architectures/)). There are two demos:

- **`sqlalchemy_quickstart`**: the `User` and `Address` models of SQLAlchemy's
  [ORM Quick Start](https://docs.sqlalchemy.org/en/20/orm/quickstart.html), with the same
  window on a sync `Session` or on the [asyncio extension](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html).
- **`sqlalchemy_contacts`**: 20,000 rows in a virtual table, with search and sorting in SQL.

## The Quick Start in a window

(Image: Users and their email addresses from SQLAlchemy's Quick Start)

1. **Declare the models** exactly as the Quick Start does, with `DeclarativeBase`, `Mapped` and
   `relationship`. The cascade deletes a user's addresses with the user:

```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. **Create an Engine and the tables:**

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

3. **Open a Session per unit of work.** Each backend function opens its own `Session`, which makes
   it safe to run in any worker thread. `selectinload` loads the addresses before the session
   closes, so the window can read `user.addresses` later:

```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. **Call it from the window with `Command.perform`.** The function runs in a thread and `then`
   receives the result on the UI thread:

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

### The asyncio version

`async_backend.py` has the same functions written with `create_async_engine`, an
`async_sessionmaker` and the aiosqlite driver. `Command.perform` sees that they are coroutine
functions and runs them on Pyzed's asyncio loop instead of a thread, so `main.py` doesn't change:

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

With `AsyncSession`, lazy loading would need an `await`, so relationships are loaded eagerly
with `selectinload`, and `expire_on_commit=False` keeps the objects readable after `commit()`.
Install `sqlalchemy[asyncio]` and `aiosqlite`; the extra brings `greenlet`, which the extension needs.

| | `Session` | `AsyncSession` |
| --- | --- | --- |
| Where it runs | A worker thread per call | Pyzed's asyncio loop |
| Pyzed call | `Command.perform(sync_function, then)` | `Command.perform(async_function, then)` or return the coroutine |
| Use it when | Most desktop apps; any driver | You already use asyncio drivers (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()
```

## Many rows: the contacts demo

A contacts manager on SQLAlchemy 2.0 and SQLite. The table holds 20,000 rows and scrolls
smoothly because `ui.table` only builds the rows in view; search and sorting run in SQL, in a
worker thread, so the window never waits for the database.

(Image: A contacts table with 20,000 rows, statistics and an edit panel)

The demo keeps the database code in `db.py`, with no Pyzed in it, and the window in `main.py`.
That split is worth copying: the same `db.py` could serve a FastAPI app or a CLI.

1. **Models are plain SQLAlchemy 2.0.**

```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. **Each query is a function with its own session.** SQLAlchemy sessions are not shared
   between threads, so every function opens one and closes it. The loaded objects keep their
   values after the session closes, which is all the view needs:

```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. **Run it with `Command.perform`.** The function runs in a thread and `then` gets the
   result on the UI thread. Returning a list of commands runs them together:

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

   Copy the state you need into locals first (`query, sort, ...`): the lambda runs later, in
   another thread, and should not read `self` while the user keeps typing.

4. **Hand all the rows to `ui.table`.** It is virtualized, so 20,000 rows cost the same as 20.
   Clicking a header calls `on_sort`, which here re-runs the query with a new `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. **Turn database errors into field errors.** `save` returns either the contact or
   `{field: message}`. A unique constraint becomes a message under the Email field:

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

Prefer `asyncio`? SQLAlchemy's `AsyncSession` with `aiosqlite` or `asyncpg` works too: return the
coroutine instead of using `Command.perform`. See
[Other libraries](https://albakore.github.io/pyzed/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()
```