Ir al contenido
Pyzedv1.9.0

FastAPI

La API de esta demo es el tutorial SQL (Relational) 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 en las que hay una API.

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

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

Sección titulada «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:

    Windows, macOS y Linux
    cd examples/integrations/fastapi_heroes
    uv add "fastapi[standard]" sqlmodel httpx   # en tu propio proyecto

  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:

    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:

    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:

    @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:

    Windows, macOS y Linux
    uv run fastapi dev main.py

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

El cliente de escritorio

Sección titulada «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:

    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:

    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, ..."}:

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

2. Cliente y servidor

Sección titulada «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:

Windows, macOS y Linux
uv run fastapi dev main.py     # terminal 1
uv run client.py               # terminal 2 (o: uv run client.py https://api.example.com)
client.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

Sección titulada «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 también funciona en un navegador.

Windows, macOS y Linux
uv run embedded.py
embedded.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

Sección titulada «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.

Windows, macOS y Linux
uv run in_process.py
in_process.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"))

Código completo

Sección titulada «Código completo»
"""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}