SQLAlchemy

SQLAlchemy runs in the same process as the window (architecture 1 in Architectures). There are two demos:

• sqlalchemy_quickstart: the User and Address models of SQLAlchemy’s ORM Quick Start, with the same window on a sync Session or on the asyncio extension.
• sqlalchemy_contacts: 20,000 rows in a virtual table, with search and sorting in SQL.

The Quick Start in a window

uv

Windows, macOS and Linux

cd examples/integrations/sqlalchemy_quickstart
uv run main.py            # Session in worker threads
uv run main.py --async    # AsyncSession with aiosqlite on Pyzed's loop

pip

macOS and Linux

cd examples/integrations/sqlalchemy_quickstart
python3 -m venv .venv
source .venv/bin/activate
pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
python main.py            # Session in worker threads
python main.py --async    # AsyncSession with aiosqlite on Pyzed's loop

Windows (PowerShell)

cd examples/integrations/sqlalchemy_quickstart
py -m venv .venv
.venv\Scripts\Activate.ps1
pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
python main.py            # Session in worker threads
python main.py --async    # AsyncSession with aiosqlite on Pyzed's loop

python

macOS and Linux

cd examples/integrations/sqlalchemy_quickstart
python3 -m pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
python3 main.py            # Session in worker threads
python3 main.py --async    # AsyncSession with aiosqlite on Pyzed's loop

Windows (PowerShell)

cd examples/integrations/sqlalchemy_quickstart
py -m pip install pyzed-ui "sqlalchemy[asyncio]" aiosqlite
py main.py            # Session in worker threads
py main.py --async    # AsyncSession with aiosqlite on Pyzed's loop

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:

   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:

   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:

   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:

   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:

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

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

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

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

"""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.

uv

Windows, macOS and Linux

cd examples/integrations/sqlalchemy_contacts
uv run main.py

pip

macOS and Linux

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

Windows (PowerShell)

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

python

macOS and Linux

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

Windows (PowerShell)

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

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.

   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:

   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:

   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:

   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:

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

Tip

Prefer asyncio? SQLAlchemy’s AsyncSession with aiosqlite or asyncpg works too: return the coroutine instead of using Command.perform. See Other libraries.

main.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

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