# FastAPI

> Crea la API del tutorial de SQL de FastAPI tal como está en su documentación y úsala desde una ventana de escritorio de Pyzed como servidor aparte, servidor integrado o en el proceso.

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

La API de esta demo es el tutorial [SQL (Relational) Databases](https://fastapi.tiangolo.com/tutorial/sql-databases/)
de FastAPI, copiado sin cambios: héroes guardados con SQLModel, con modelos separados para crear,
leer y actualizar. La ventana de Pyzed es un cliente de escritorio para ella, y la misma ventana
funciona en las tres [arquitecturas](https://albakore.github.io/pyzed/es/integrations/architectures/) en las que hay una API.

(Image: Una ventana de escritorio con la lista de héroes de la API de FastAPI y un formulario de edición)

```text
fastapi_heroes/
├── main.py            # la API, exactamente como en el tutorial de FastAPI
├── heroes_window.py   # la ventana de Pyzed, que solo habla HTTP (httpx)
├── client.py          # 2. cliente y servidor
├── embedded.py        # 3. servidor integrado
└── in_process.py      # 4. en el proceso
```

## 1. La API, como la enseña FastAPI

1. **Instala FastAPI con sus extras estándar**, que traen la CLI `fastapi` y Uvicorn. El
   `pyproject.toml` de la demo ya los incluye:

2. **Escribe los modelos.** `HeroBase` tiene los campos comunes, `Hero` es la tabla, `HeroPublic`
   es lo que devuelve la API (nunca el `secret_name`), y `HeroCreate` y `HeroUpdate` describen
   los cuerpos de las peticiones:

```python
class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str

class HeroPublic(HeroBase):
    id: int

class HeroCreate(HeroBase):
    secret_name: str

class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None
```

3. **Declara una dependencia de sesión** para que cada petición tenga su propia `Session`:

```python
def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]
```

4. **Escribe las operaciones de ruta.** `response_model` filtra la salida y FastAPI valida la
   entrada, y responde **422** con una entrada por cada campo inválido:

```python
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    hero_db.sqlmodel_update(hero_data)
    ...
```

5. **Ejecútala con la CLI de FastAPI** y abre la documentación interactiva en
   [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs):

   `fastapi dev` recarga con cada cambio; `fastapi run main.py` es el comando de producción.

## El cliente de escritorio

`heroes_window.py` es una `pz.App` que habla con la API con un `httpx.AsyncClient`. Recibe una
función que crea el cliente, así que la misma ventana funciona con cualquiera de las
arquitecturas de abajo.

1. **Envía las peticiones desde corrutinas.** Devolver una corrutina desde un callback la ejecuta
   en el loop de asyncio de Pyzed, y lo que devuelve llega a `update`. El cliente se crea la
   primera vez que se usa, para que pertenezca a ese loop:

```python
async def api(self, method, url, **kwargs):
    if self.client is None:
        self.client = self.make_client()
    try:
        return await self.client.request(method, url, **kwargs)
    except httpx.HTTPError as error:
        return f"Can't reach the API: {error}"

async def load(self):
    response = await self.api("GET", "/heroes/", params={"limit": 100})
    if isinstance(response, str):
        return ("offline", response)
    return ("loaded", response.json())
```

2. **Crea con POST y edita con PATCH.** Solo se envían los campos que el usuario rellenó, que es
   lo que esperan `HeroUpdate` y `exclude_unset=True`:

```python
async def save(self):
    body = {key: value for key, value in self.draft.items() if value != ""}
    if self.selected is None:
        response = await self.api("POST", "/heroes/", json=body)
    else:
        response = await self.api("PATCH", f"/heroes/{self.selected['id']}", json=body)
    if response.status_code == 422:
        return ("invalid", response.json())
    ...
```

3. **Muestra el 422 de FastAPI en cada campo.** `pz.validation_errors` lee `{"detail": [...]}` y
   devuelve `{"age": "Input should be a valid integer, ..."}`:

```python
def update(self, message):
    kind, value = message
    if kind == "invalid":
        self.errors = pz.validation_errors(value)
```

## 2. Cliente y servidor

La API y la ventana son dos programas. Así se usa una API desplegada, y así trabajas en las dos
a la vez con recarga:

`client.py`:

```py
"""Architecture 1, client and server: the API runs on its own and the window talks to it over HTTP.

    uv run fastapi dev main.py     # terminal 1: the API (FastAPI's own CLI)
    uv run client.py               # terminal 2: the desktop client
    uv run client.py https://api.example.com   # or any other deployment of main.py

The usual setup when the API also serves a website or a mobile app, or runs on a server.
"""

import sys

import httpx

import pyzed as pz

from heroes_window import HeroesWindow

API = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8000"

if __name__ == "__main__":
    pz.run(HeroesWindow(lambda: httpx.AsyncClient(base_url=API, timeout=5), mode=f"server at {API}",
                        docs_url=f"{API}/docs"))
```

## 3. Servidor integrado

Un solo comando arranca la API en un hilo en segundo plano con Uvicorn, el servidor que usa
`fastapi run`, y abre la ventana. Mientras la ventana está abierta,
[http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs) también funciona en un navegador.

`embedded.py`:

```py
"""Architecture 2, embedded server: one program starts the API and opens the window.

    uv run embedded.py

The API runs with Uvicorn (the server `fastapi run` and `fastapi dev` use) in a background
thread, so other programs on this computer, a browser at /docs for instance, can use it
while the window is open. It stops when the window closes.
"""

import threading

import httpx
import uvicorn

import pyzed as pz

from heroes_window import HeroesWindow
from main import app

HOST, PORT = "127.0.0.1", 8000

def start_api() -> None:
    # What `fastapi run main.py` does, but in a thread of this process.
    server = uvicorn.Server(uvicorn.Config(app, host=HOST, port=PORT, log_level="warning"))
    threading.Thread(target=server.run, daemon=True).start()

if __name__ == "__main__":
    start_api()
    url = f"http://{HOST}:{PORT}"
    pz.run(HeroesWindow(lambda: httpx.AsyncClient(base_url=url, timeout=5), mode=f"embedded at {url}",
                        docs_url=f"{url}/docs"))
```

## 4. En el proceso

Sin servidor y sin puerto: `httpx.ASGITransport` entrega cada petición a la app de FastAPI como
una llamada a función. El programa funciona sin conexión, y el código de la API no cambia.

`in_process.py`:

```py
"""Architecture 3, in process: the window calls the FastAPI app directly, with no server or port.

    uv run in_process.py

httpx can send requests straight to an ASGI app (httpx.ASGITransport), so the same API code
and the same client code run inside one desktop program that works offline. Nothing listens
on the network. ASGITransport doesn't send startup events, so the tables are created here.
"""

import httpx

import pyzed as pz

from heroes_window import HeroesWindow
from main import app, create_db_and_tables

if __name__ == "__main__":
    create_db_and_tables()
    transport = httpx.ASGITransport(app=app)
    pz.run(HeroesWindow(lambda: httpx.AsyncClient(transport=transport, base_url="http://heroes"), mode="in process"))
```

**Caution:**

`ASGITransport` no envía los eventos de startup ni de lifespan, así que el código de
`@app.on_event("startup")` o de una función `lifespan` no se ejecuta. `in_process.py` llama a
`create_db_and_tables()` por su cuenta.

## Código completo

**main.py (API):**
```py
"""The FastAPI app, exactly as FastAPI's "SQL (Relational) Databases" tutorial writes it:
https://fastapi.tiangolo.com/tutorial/sql-databases/

    uv run fastapi dev main.py      # http://127.0.0.1:8000, interactive docs at /docs

Nothing in this file knows about Pyzed. The three desktop entry points (client.py,
embedded.py, in_process.py) reuse it unchanged.
"""

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str

class HeroPublic(HeroBase):
    id: int

class HeroCreate(HeroBase):
    secret_name: str

class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    hero_db.sqlmodel_update(hero_data)
    session.add(hero_db)
    session.commit()
    session.refresh(hero_db)
    return hero_db

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}
```

**heroes_window.py:**
```py
"""The Pyzed window: a desktop client for the heroes API in main.py, written with httpx.

It only knows the API's URLs and JSON, so it works the same whether the API runs in another
process, in a thread of this one or in-process without a network (see client.py,
embedded.py and in_process.py).
"""

from __future__ import annotations

import webbrowser
from typing import Callable

import httpx

import pyzed as pz
from pyzed import ui

EMPTY = {"name": "", "secret_name": "", "age": ""}

class HeroesWindow(pz.App):
    title = "Heroes · FastAPI"
    size = (980, 620)

    def __init__(self, make_client: Callable[[], httpx.AsyncClient], mode: str, docs_url: str | None = None):
        self.make_client = make_client  # called once, inside a coroutine, so the client lives on Pyzed's loop
        self.client: httpx.AsyncClient | None = None
        self.mode, self.docs_url = mode, docs_url
        self.heroes: list[dict] = []
        self.selected: dict | None = None  # the hero being edited, or None for a new one
        self.draft = dict(EMPTY)
        self.errors: dict[str, str] = {}
        self.status = "Connecting…"

    # -- the API, with httpx --------------------------------------------------------------
    async def api(self, method: str, url: str, **kwargs) -> httpx.Response | str:
        """Sends a request; returns the response, or the error text when the API can't be reached."""
        if self.client is None:
            self.client = self.make_client()
        try:
            return await self.client.request(method, url, **kwargs)
        except httpx.HTTPError as error:
            return f"Can't reach the API: {error or type(error).__name__}"

    async def load(self):
        response = await self.api("GET", "/heroes/", params={"limit": 100})  # GET /heroes/?offset=0&limit=100
        if isinstance(response, str):
            return ("offline", response)
        return ("loaded", response.json())

    async def save(self):
        body = {key: value for key, value in self.draft.items() if value != ""}  # leave empty fields out
        if self.selected is None:
            response = await self.api("POST", "/heroes/", json=body)
        else:
            response = await self.api("PATCH", f"/heroes/{self.selected['id']}", json=body)
        if isinstance(response, str):
            return ("offline", response)
        if response.status_code == 422:  # FastAPI's validation error, one entry per field
            return ("invalid", response.json())
        response.raise_for_status()
        return ("saved", response.json())

    async def delete(self):
        response = await self.api("DELETE", f"/heroes/{self.selected['id']}")
        if isinstance(response, str):
            return ("offline", response)
        return ("deleted", self.selected["id"])

    # -- state -----------------------------------------------------------------------------
    def on_start(self):
        return self.load()

    def pick(self, hero: dict | None):
        self.selected, self.errors = hero, {}
        self.draft = dict(EMPTY) if hero is None else {
            "name": hero["name"], "secret_name": "", "age": "" if hero["age"] is None else str(hero["age"])}

    def edit(self, key: str):
        return lambda value: self.draft.__setitem__(key, value)

    def update(self, message):
        kind, value = message
        if kind == "loaded":
            self.heroes, self.status = value, f"{len(value)} heroes · {self.mode}"
        elif kind == "offline":
            self.status = value
        elif kind == "invalid":
            self.errors = pz.validation_errors(value)
        elif kind == "saved":
            self.pick(value)
            return self.load()
        elif kind == "deleted":
            self.pick(None)
            return self.load()

    # -- view --------------------------------------------------------------------------------
    def form(self):
        e = self.errors.get
        new = self.selected is None
        return pz.column([
            ui.heading("New hero" if new else f"Hero #{self.selected['id']}", level=3),
            ui.field("Name", ui.input(self.draft["name"], self.edit("name"), placeholder="Deadpond", width=220),
                     error=e("name")),
            ui.field("Secret name", ui.input(self.draft["secret_name"], self.edit("secret_name"),
                                             placeholder="Dive Wilson" if new else "unchanged", width=220),
                     description="Never sent back by the API", error=e("secret_name")),
            ui.field("Age", ui.input(self.draft["age"], self.edit("age"), placeholder="optional", width=220),
                     error=e("age")),
            pz.row([
                None if new else ui.button("Delete", self.delete, icon="trash-2", variant="ghost", tint="error"),
                pz.hspace(),
                ui.button("New", lambda: self.pick(None), variant="ghost") if not new else None,
                ui.button("Create" if new else "Save", self.save, icon="check", variant="solid"),
            ], spacing=8),
        ], spacing=14, padding=16, width=380)

    def view(self):
        t = self.resolved_theme()
        columns = [ui.Column("ID", "id", width=50), ui.Column("Name", "name"),
                   ui.Column("Age", lambda h: "—" if h["age"] is None else h["age"], width=70, align="end")]
        table = ui.table(columns, self.heroes, on_row_press=self.pick, selected=self.selected, height="fill")
        offline = self.status.startswith("Can't")
        status = ui.status_bar(
            [ui.status_item(self.status, icon="cloud-off" if offline else "server",
                            icon_color="error" if offline else "success")],
            [ui.status_item("API docs", lambda: webbrowser.open(self.docs_url), icon="book-open")] if self.docs_url else [],
        )
        body = pz.row([pz.container(table, padding=16, width="fill", height="fill"),
                       pz.container(self.form(), height="fill", bg=t.panel)])
        return pz.container(pz.column([body, status]), width="fill", height="fill", bg=t.editor)
```