DEV Community: Martin Palopoli

SMTP outbound as a first-class citizen: send emails without pip install yagmail

Martin Palopoli — Mon, 13 Jul 2026 17:43:42 +0000

Notifications, password resets, magic links, alerts. Every language solves SMTP outbound with an external library. In Fitz, smtp.send(opts) is a language builtin. Async from day one. Bit-for-bit parity between fitz run and fitz build. No pip install yagmail / npm install nodemailer / cargo add lettre / Maven JavaMail.

The detail that gets forgotten

Any serious application eventually needs to send emails:

Notifications when a service goes down (incidents → on-call email).
Password reset / magic-link auth (unique link to the user's email).
Welcome email after a signup.
Daily / weekly digest reports dispatched from a cron job.
Alerts when a nightly job fails.

And every language solves this with an external library.

We hit the gap while building fitzwatch (open-source status page written in pure Fitz): the heart of the product alerts on-call when a monitor goes down. Without native SMTP, the only option was webhook → n8n/zapier → email. Two external systems just to send one email.

Today, the 8 sub-blocks of the mini-release are in main. One line — smtp.send({...}).await? — and the email goes out.

The typical Python stack

pip install yagmail
# or use stdlib smtplib (RFC 5321 by hand)

# With yagmail (friendly syntax):
import yagmail

yag = yagmail.SMTP("user@example.com", "password")
yag.send("dest@example.com", "Subject", "Body text")

# With stdlib smtplib (low-level):
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

msg = MIMEMultipart("alternative")
msg["Subject"] = "Subject"
msg["From"] = "user@example.com"
msg["To"] = "dest@example.com"
msg.attach(MIMEText("Body text", "plain"))
msg.attach(MIMEText("<p>Body HTML</p>", "html"))

with smtplib.SMTP("smtp.gmail.com", 587) as server:
    server.starttls()
    server.login("user@example.com", "password")
    server.send_message(msg)

Two options, both require reading the RFC or learning the yagmail API.

The typical JS/Node stack

npm install nodemailer

import nodemailer from "nodemailer"

const transporter = nodemailer.createTransport({
    host: "smtp.gmail.com",
    port: 587,
    secure: false,
    auth: { user: "user@example.com", pass: "password" },
})

await transporter.sendMail({
    from: "user@example.com",
    to: "dest@example.com",
    subject: "Subject",
    text: "Body text",
    html: "<p>Body HTML</p>",
})

One library, well-designed, but npm install brings ~50 transitive packages to package.json.

The typical Rust stack

cargo add lettre
cargo add tokio --features full

use lettre::message::{header::ContentType, Mailbox, MultiPart, SinglePart};
use lettre::transport::smtp::authentication::Credentials;
use lettre::{AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor};

let creds = Credentials::new("user@example.com".into(), "password".into());
let mailer = AsyncSmtpTransport::<Tokio1Executor>::starttls_relay("smtp.gmail.com")
    .unwrap()
    .port(587)
    .credentials(creds)
    .build();

let email = Message::builder()
    .from("user@example.com".parse::<Mailbox>().unwrap())
    .to("dest@example.com".parse::<Mailbox>().unwrap())
    .subject("Subject")
    .multipart(MultiPart::alternative()
        .singlepart(SinglePart::builder().header(ContentType::TEXT_PLAIN).body("Body text".to_string()))
        .singlepart(SinglePart::builder().header(ContentType::TEXT_HTML).body("<p>Body HTML</p>".to_string()))
    )
    .unwrap();

mailer.send(email).await.unwrap();

Functional, but verbose. 25 lines to send ONE email.

The typical Java stack

<dependency>
    <groupId>com.sun.mail</groupId>
    <artifactId>jakarta.mail</artifactId>
    <version>2.0.1</version>
</dependency>

And then the canonical recipe of Properties + Session + MimeMessage + Transport.send(...) that probably adds 40 lines.

The Fitz stack

let r = smtp.send({
    "to": "dest@example.com",
    "from": "user@example.com",
    "subject": "Subject",
    "body_text": "Body text",
    "body_html": "<p>Body HTML</p>",
}).await?

That's it.

smtp is a language builtin. No pip install/npm install/cargo add.
body_text + body_html together → automatic multipart/alternative.
.await? propagates errors as Result::Err(Str). The compiler enforces handling.
Config (host, port, user, password, TLS) is read from env vars at first send. Same convention any app in Kubernetes/Docker expects.

r.delivered, r.message_id, and r.duration_ms come out typed as Bool/Str/Int.

What `smtp.send` guarantees you

1. Zero external deps

The fitz binary ships with lettre 0.11 statically linked inside. When you fitz build, the result is ONE executable of ~10 MB containing the SMTP client, TLS via rustls, no openssl on the target host.

ldd ./my-app
# linux-vdso.so.1
# libgcc_s.so.1
# libc.so.6
# (no smtp libs, no openssl)

2. Bit-for-bit parity `fitz run` ↔ `fitz build`

The interpreter (fitz run) and codegen to Rust (fitz build) emit the same wire SMTP. Same handshake, same SCRAM auth, same Message-ID. Bug in one path = bug in the other. E2E tests validate this on every commit.

3. Async from day one

smtp.send(...) returns Future<Result<SmtpResult>>. Integrates naturally with the rest of the stack:

@background
async fn send_welcome(email: Str) -> Null {
    let _ = smtp.send({
        "to": email,
        "subject": "Welcome",
        "body": "Hi, thanks for joining.",
    }).await
    return null
}

@post("/signup")
fn signup(input: SignupInput) {
    // ... create the user in the DB ...
    let _ = spawn(send_welcome(input.email))
    return 201 { "id": new_user_id }
}

The handler returns 201 to the client immediately. The email goes out in another tokio task. If the SMTP server takes 2 seconds, the client doesn't care — it already responded 2 seconds ago.

And with @cron you build digests:

@cron("0 0 9 * * *")  // every day at 09:00
async fn daily_digest() -> Null {
    let r = smtp.send({
        "to": "team@example.com",
        "subject": "Daily digest",
        "body_html": "<p>Today: ...</p>",
    }).await
    match r {
        Ok(_) => log.info("digest.sent"),
        Err(e) => log.error("digest.failed", error: e),
    }
    return null
}

4. `Result<T>` as error model

Transport errors (DNS, auth, TLS, server reject) come as Result::Err(Str) with "smtp: " prefix:

match smtp.send(opts).await {
    Ok(r) => log.info("smtp.delivered", message_id: r.message_id),
    Err(e) => {
        // The prefix lets you classify:
        // "smtp: server rejected mail: ..."     → 5xx from server
        // "smtp: transient error: ..."          → 4xx temporary
        // "smtp: client error: ..."             → TLS/auth fail
        // "smtp: invalid `to` address ..."      → parse error
        log.warn("smtp.failed", error: e)
    }
}

The static checker enforces handling. If your fn returns Result<...>, you can propagate with ?. If not, the checker demands match. Zero runtime exceptions.

5. Magic-link auth in 30 lines

The showcase case of Fitz's first-class web stack: HTTP server-side + auth with jwt.encode + SMTP outbound, all combined in one binary.

type EmailRequest {
    email: Str
}

@post("/auth/magic-link")
async fn magic_link(input: EmailRequest) -> Result<Str> {
    let payload = { "email": input.email }
    let token = jwt.encode(payload, "secret")
    let link = "https://app.example.com/verify?t={token}"
    let r = smtp.send({
        "to": input.email,
        "subject": "Your login link",
        "body_text": "Click here:\n{link}\n\nExpires in 5 min.",
        "body_html": "<p>Click <a href=\"{link}\">here</a>.</p>",
    }).await?
    return Ok(r.message_id)
}

No Auth0, no Supabase, no Stripe webhooks. One binary, one port, one command to deploy.

Local setup with MailHog

For dev you don't want to send real emails. MailHog is a fake SMTP server with a web UI that retains emails without forwarding:

docker run -d --name mailhog -p 1025:1025 -p 8025:8025 mailhog/mailhog
export SMTP_HOST=localhost
export SMTP_PORT=1025
export SMTP_TLS=none
fitz run app.fitz
# Then open http://localhost:8025 in the browser.

In production you change the env vars to Gmail/SES/SendGrid/Mailgun/whatever — your Fitz code doesn't change.

Honest trade-offs (MVP)

No attachments yet. The attachments key is rejected with a clear error. Workaround: bundle attachments via S3/object storage + link in the body. The cleanest pattern for serious production anyway.
No programmatic smtp.configure(...). Config only comes from env vars today. If you need multi-tenant with a different SMTP per tenant, that's pending as post-MVP debt.
No built-in retry. If smtp.send fails, the caller decides what to do. The canonical pattern for production is outbox table + @cron with retry.

This is NOT FUD — these are explicit MVP decisions. All 3 are refinable if real demand shows up.

Closing

smtp.send isn't "another easier library". It's the same mental model you already use elsewhere in Fitz: Result<T> as return type, ? to propagate, .await for async, @background + spawn for fire-and-forget.

If you already know how to chain http.get(...).await? or db.query(...).await?, you already know how to send emails. Zero new API to learn.

And when fitzwatch resumes, the line that was pending to notify incidents goes from "webhook to n8n + translation to SMTP" to smtp.send({...}).await?. One line. That's Fitz.

Try it: fitz version 0.18.0+ (already in GitHub releases). Exhaustive chapter + 3 runnable examples against MailHog in the "SMTP outbound" subsection of chapter 17 of the guide.

Stay tuned: the Fitz series publishes on dev.to every time we close a major mini-release. The next one will cover a concrete feature detected during fitzwatch development.

SMTP outbound como ciudadano de primera clase: mandá emails sin pip install yagmail

Martin Palopoli — Mon, 13 Jul 2026 17:43:20 +0000

Notificaciones, password resets, magic links, alerts. Todos los lenguajes resuelven SMTP outbound con una librería externa. En Fitz, smtp.send(opts) es builtin del lenguaje. Async desde día uno. Paridad bit-a-bit fitz run ↔ fitz build. Sin pip install yagmail / npm install nodemailer / cargo add lettre / Maven JavaMail.

El detalle que se olvida

Cualquier aplicación seria termina necesitando mandar mails:

Notificación cuando se cae un servicio (incidents → email al on-call).
Password reset / magic-link auth (link único al email del user).
Welcome email después de un signup.
Reportes diarios / weekly digests despachados desde un cron job.
Alertas de jobs nocturnos que fallan.

Y todos los lenguajes lo resuelven con una librería externa.

Detectamos el gap construyendo fitzwatch (status page open-source escrito en Fitz puro): el corazón del producto avisa al on-call cuando un monitor cae. Sin SMTP nativo, la única opción era webhook → n8n/zapier → email. Dos sistemas externos para mandar un email.

Hoy, los 8 sub-bloques de la mini-tanda están en main. Una línea — smtp.send({...}).await? — y el email sale.

El stack típico de Python

pip install yagmail
# o usar smtplib del stdlib (RFC 5321 a mano)

# Con yagmail (sintaxis amigable):
import yagmail

yag = yagmail.SMTP("user@example.com", "password")
yag.send("dest@example.com", "Subject", "Body text")

# Con smtplib stdlib (bajo nivel):
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

msg = MIMEMultipart("alternative")
msg["Subject"] = "Subject"
msg["From"] = "user@example.com"
msg["To"] = "dest@example.com"
msg.attach(MIMEText("Body text", "plain"))
msg.attach(MIMEText("<p>Body HTML</p>", "html"))

with smtplib.SMTP("smtp.gmail.com", 587) as server:
    server.starttls()
    server.login("user@example.com", "password")
    server.send_message(msg)

Dos opciones, ambas requieren leerte la RFC o aprender la API de yagmail.

El stack típico de JS/Node

npm install nodemailer

import nodemailer from "nodemailer"

const transporter = nodemailer.createTransport({
    host: "smtp.gmail.com",
    port: 587,
    secure: false,
    auth: { user: "user@example.com", pass: "password" },
})

await transporter.sendMail({
    from: "user@example.com",
    to: "dest@example.com",
    subject: "Subject",
    text: "Body text",
    html: "<p>Body HTML</p>",
})

Una sola lib, bien diseñada, pero npm install agrega ~50 paquetes transitivos al package.json.

El stack típico de Rust

cargo add lettre
cargo add tokio --features full

use lettre::message::{header::ContentType, Mailbox, MultiPart, SinglePart};
use lettre::transport::smtp::authentication::Credentials;
use lettre::{AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor};

let creds = Credentials::new("user@example.com".into(), "password".into());
let mailer = AsyncSmtpTransport::<Tokio1Executor>::starttls_relay("smtp.gmail.com")
    .unwrap()
    .port(587)
    .credentials(creds)
    .build();

let email = Message::builder()
    .from("user@example.com".parse::<Mailbox>().unwrap())
    .to("dest@example.com".parse::<Mailbox>().unwrap())
    .subject("Subject")
    .multipart(MultiPart::alternative()
        .singlepart(SinglePart::builder().header(ContentType::TEXT_PLAIN).body("Body text".to_string()))
        .singlepart(SinglePart::builder().header(ContentType::TEXT_HTML).body("<p>Body HTML</p>".to_string()))
    )
    .unwrap();

mailer.send(email).await.unwrap();

Funcional, pero verboso. 25 líneas para mandar UN mail.

El stack típico de Java

<dependency>
    <groupId>com.sun.mail</groupId>
    <artifactId>jakarta.mail</artifactId>
    <version>2.0.1</version>
</dependency>

Y después la receta canónica de Properties + Session + MimeMessage + Transport.send(...) que probablemente sumás 40 líneas.

El stack de Fitz

let r = smtp.send({
    "to": "dest@example.com",
    "from": "user@example.com",
    "subject": "Subject",
    "body_text": "Body text",
    "body_html": "<p>Body HTML</p>",
}).await?

Que es esto.

smtp es builtin del lenguaje. No hay pip install/npm install/cargo add.
body_text + body_html juntos → multipart/alternative automático.
.await? propaga errores como Result::Err(Str). El compilador exige manejo.
Config (host, port, user, password, TLS) se lee de env vars al primer send. Lo mismo que cualquier app en Kubernetes/Docker espera.

r.delivered, r.message_id y r.duration_ms salen tipados como Bool/Str/Int.

Lo que `smtp.send` te garantiza

1. Cero deps externas

El binario fitz ya viene con lettre 0.11 linkeado estático adentro. Cuando hacés fitz build, el resultado es UN ejecutable de ~10 MB que tiene cliente SMTP, TLS via rustls, sin openssl en el host destino.

ldd ./mi-app
# linux-vdso.so.1
# libgcc_s.so.1
# libc.so.6
# (nada de smtp libs ni openssl)

2. Paridad bit-a-bit `fitz run` ↔ `fitz build`

El intérprete (fitz run) y el codegen a Rust (fitz build) emiten el mismo wire SMTP. Mismo handshake, mismo SCRAM auth, mismo Message-ID. Bug en un path = bug en el otro. Tests E2E lo validan en cada commit.

3. Async desde día uno

smtp.send(...) devuelve Future<Result<SmtpResult>>. Integra natural con el resto del stack:

@background
async fn send_welcome(email: Str) -> Null {
    let _ = smtp.send({
        "to": email,
        "subject": "Bienvenido",
        "body": "Hola, gracias por unirte.",
    }).await
    return null
}

@post("/signup")
fn signup(input: SignupInput) {
    // ... crear el user en la DB ...
    let _ = spawn(send_welcome(input.email))
    return 201 { "id": new_user_id }
}

El handler responde 201 al cliente inmediato. El email se manda en otro task tokio. Si el SMTP server tarda 2 segundos, no le importa al cliente — ya respondió hace 2 segundos.

Y con @cron armás digests:

@cron("0 0 9 * * *")  // todos los días 09:00
async fn daily_digest() -> Null {
    let r = smtp.send({
        "to": "team@example.com",
        "subject": "Daily digest",
        "body_html": "<p>Hoy: ...</p>",
    }).await
    match r {
        Ok(_) => log.info("digest.sent"),
        Err(e) => log.error("digest.failed", error: e),
    }
    return null
}

4. `Result<T>` como modelo de error

Errores de transporte (DNS, auth, TLS, server reject) llegan como Result::Err(Str) con prefijo "smtp: ":

match smtp.send(opts).await {
    Ok(r) => log.info("smtp.delivered", message_id: r.message_id),
    Err(e) => {
        // El prefijo te permite clasificar:
        // "smtp: server rejected mail: ..."     → 5xx del server
        // "smtp: transient error: ..."          → 4xx temporario
        // "smtp: client error: ..."             → TLS/auth fail
        // "smtp: invalid `to` address ..."      → parse error
        log.warn("smtp.failed", error: e)
    }
}

El checker estático exige manejo. Si tu fn retorna Result<...>, podés propagar con ?. Si no, te exige match. Cero excepciones runtime.

5. Magic-link auth en 30 líneas

El caso showcase del stack web first-class de Fitz: HTTP server-side + auth con jwt.encode + SMTP outbound, todo combinado en un binario.

type EmailRequest {
    email: Str
}

@post("/auth/magic-link")
async fn magic_link(input: EmailRequest) -> Result<Str> {
    let payload = { "email": input.email }
    let token = jwt.encode(payload, "secret")
    let link = "https://app.example.com/verify?t={token}"
    let r = smtp.send({
        "to": input.email,
        "subject": "Tu link de login",
        "body_text": "Hacé click acá:\n{link}\n\nExpira en 5 min.",
        "body_html": "<p>Hacé click <a href=\"{link}\">acá</a>.</p>",
    }).await?
    return Ok(r.message_id)
}

Sin Auth0, sin Supabase, sin Stripe webhooks. Un binario, un puerto, un comando para deployarlo.

Setup local con MailHog

Para dev no querés mandar mails reales. MailHog es un SMTP server fake con UI web que retiene los mails sin reenviar:

docker run -d --name mailhog -p 1025:1025 -p 8025:8025 mailhog/mailhog
export SMTP_HOST=localhost
export SMTP_PORT=1025
export SMTP_TLS=none
fitz run app.fitz
# Después abrís http://localhost:8025 en el browser.

En producción cambiás las env vars a Gmail/SES/SendGrid/Mailgun/lo que sea — el código de Fitz no cambia.

Trade-offs honestos (MVP)

Sin attachments todavía. La key attachments se rechaza con error claro. Workaround: bundle attachments via S3/object storage + link en el body. Lo más limpio para producción serio igual.
Sin smtp.configure(...) programático. Config sale solo de env vars hoy. Si necesitás multi-tenant con un SMTP por tenant, te queda como deuda post-MVP.
Sin retry built-in. Si smtp.send falla, el caller decide qué hacer. El patrón canónico para producción es tabla de outbox + @cron con retry.

Esto NO es FUD — son decisiones explícitas del MVP. Los 3 son refinables si entra demanda real.

Cierre

smtp.send no es "una librería más fácil". Es el mismo modelo mental que ya usás en el resto de Fitz: Result<T> como retorno, ? para propagar, .await para async, @background + spawn para fire-and-forget.

Si ya sabés cómo encadenar http.get(...).await? o db.query(...).await?, ya sabés cómo mandar mails. Cero API nuevo que aprender.

Y cuando fitzwatch reanude, la línea que tenía pendiente para notificar incidents pasa de "webhook a n8n + traducción a SMTP" a smtp.send({...}).await?. Una línea. Eso es Fitz.

Probalo: fitz versión 0.18.0+ (ya en GitHub releases). Cap exhaustivo + 3 ejemplos runnable contra MailHog en la sub-sección "SMTP outbound" del cap 17 de la guía.

Mantenete al día: la serie Fitz se publica en dev.to cada vez que cerramos una mini-tanda grande. La próxima cubrirá una feature concreta detectada durante fitzwatch.

HTTP client as a first-class citizen: zero deps, bit-for-bit parity, async by default

Martin Palopoli — Wed, 08 Jul 2026 10:05:02 +0000

Webhooks, scraping, health checks, proxying. Every language solves outbound HTTP with an external library. In Fitz, http.get/post/put/delete/head/request are language builtins. Async from day one. Bit-for-bit parity between the interpreter and the native binary. No pip install requests / npm install axios / cargo add reqwest required.

The forgotten detail

Fitz has had @get/@post since Phase 4. Serving HTTP is trivial. But until last week, if your handler needed to call another service (Slack webhook, scraping, health check, proxying to an upstream), you had two options:

Import Python with from python import urllib.request and bundle CPython (~22 MB extra to the binary).
Don't do it.

Any serious project eventually needs this. I caught the gap building fitzwatch (open-source status page written in pure Fitz): the heart of the product is http.head(monitor.target) every N seconds. Without that line, the language didn't reach.

A week later, the 6 builtins are in main, with bit-for-bit parity between fitz run and fitz build, static validation via the checker, and zero external deps in the final binary.

The typical Python stack

pip install requests
# or pip install httpx (if you want async)

import requests

r = requests.get("https://api.example.com/users/42", timeout=5)
if r.status_code == 200:
    user = r.json()
    print(user)
elif r.status_code == 404:
    print("not found")
else:
    raise Exception(f"unexpected status: {r.status_code}")

# Async version with httpx
import httpx
async with httpx.AsyncClient() as client:
    r = await client.get("https://api.example.com/users/42", timeout=5)
    user = r.json()

Two libraries to cover sync and async. requests (sync, no asyncio integration) or httpx (async, different API).

The typical JS/TS stack

npm install axios
# or just fetch from std if Node 18+

import axios from "axios"

const r = await axios.get("https://api.example.com/users/42", { timeout: 5000 })
if (r.status === 200) {
    console.log(r.data)
} else if (r.status === 404) {
    console.log("not found")
} else {
    throw new Error(`unexpected status: ${r.status}`)
}

axios throws on any status >= 400 by default — either disable it or wrap in try/catch.

The typical Rust stack

# Cargo.toml
[dependencies]
reqwest = { version = "0.12", features = ["json"] }
tokio = { version = "1", features = ["full"] }
serde_json = "1"

let client = reqwest::Client::new();
let r = client
    .get("https://api.example.com/users/42")
    .timeout(Duration::from_secs(5))
    .send()
    .await?;

match r.status().as_u16() {
    200 => {
        let user: serde_json::Value = r.json().await?;
        println!("{}", user);
    }
    404 => println!("not found"),
    other => eprintln!("unexpected status: {}", other),
}

Three deps, two crates to import, a Client to construct. Works very well — reqwest is excellent — but it's a separate world from std.

Same thing in Fitz

async fn run() -> Result<Null> {
    let r = http.get("https://api.example.com/users/42").await?
    match r.status {
        200 => print("user: {r.body}"),
        404 => print("not found"),
        s => print("unexpected status: {s}"),
    }
    return Ok(null)
}

// Top-level
match run().await {
    Ok(_) => print(""),
    Err(e) => print("transport error: {e}"),
}

No pip install. No npm install. No cargo add. The http module is part of the fitz binary.

The raw table

Piece	Python (`requests`/`httpx`)	JS (`axios`)	Rust (`reqwest`)	Fitz
How you get it	`pip install`	`npm install`	`cargo add`	built-in
Sync + Async	2 different libs	1 lib	1 lib	1 async builtin
4xx/5xx as error	depends (axios yes, requests no)	yes by default	no,`Result<Response>` only for transport	no,`r.status`
Transport errors	exception	exception	`Result::Err`	`Result::Err(Str)`
JSON body auto	manual `.json()`	automatic in `r.data`	manual `.json::<T>().await?`	pass a `Map<Str, Any>` and it gets serialized
`Content-Type: application/json` auto on send	manual with `json=`	yes in axios	manual with `.json(&v)`	yes when body is `Map`
TLS	depends on OS / OpenSSL	native	feature flag	static `rustls-tls`, no OpenSSL on host
Integrated with OpenAPI schema	not applicable	not applicable	not applicable	uses the same `type`s as server-side

The 6 builtins, all on one page

// GET without body
let r = http.get("https://example.com/api").await?

// HEAD — useful for health checks (no body download)
let r = http.head("https://example.com/healthz").await?

// DELETE without body
let r = http.delete("https://example.com/api/items/42").await?

// POST with Map body (auto-JSON + Content-Type)
let r = http.post(
    "https://example.com/api/items",
    {"name": "ada", "role": "admin"}
).await?

// PUT with Str body (no header changes)
let r = http.put("https://example.com/api/raw", "hello world").await?

// POST with Bytes body
let bin: Bytes = bytes([0x89, 0x50, 0x4e, 0x47])  // PNG signature
let r = http.post("https://example.com/upload", bin).await?

// request — low-level with options
let r = http.request({
    "method": "PATCH",
    "url": "https://example.com/api/items/42",
    "timeout_ms": 5000,
    "headers": {"X-Token": "abc"},
    "body": {"status": "active"},
    "follow_redirects": false,
}).await?

// Response: 4 typed fields
print("status      = {r.status}")
print("body        = {r.body}")
print("duration_ms = {r.duration_ms}")
let server: Str = match r.headers.get("server") {
    Ok(v) => v,
    Err(_) => "(unknown)",
}
print("server      = {server}")

r is always an HttpClientResponse { status: Int, body: Str, headers: Map<Str, Str>, duration_ms: Int }. The type is a language built-in, parallel to Request/Response from server-side HTTP.

Error handling: the important delta

We made a different choice than the neighbor stacks: 4xx/5xx statuses are NOT errors. If the server responds 404 or 500, you get it as Ok(response) with r.status set. Only transport errors (DNS, timeout, TLS handshake, broken connection) drop into Result::Err(Str).

Why? Because 4xx/5xx have useful bodies. If your API responds 422 with {"error": "validation failed", "field": "email"}, you want to read that. Making it a throw / Err loses the body or forces you into a try/catch that catches too much.

let r = http.post("https://api.example.com/users", {"email": "bad"}).await?

if (r.status >= 400 and r.status < 500) {
    // Your API is telling you something. Read it.
    print("client error: {r.body}")
} else if (r.status >= 500) {
    print("server crashed: {r.status}")
} else {
    print("OK: {r.body}")
}

Errs only show up when there was no response:

match http.get("https://does-not-exist.example").await {
    Ok(r)  => print("status: {r.status}"),
    Err(e) => print("transport: {e}"),
    // → "transport: http: error sending request..."
}

match http.request({"method": "GET", "url": "https://slow.example", "timeout_ms": 100}).await {
    Ok(r)  => print("reached"),
    Err(e) => print("timeout: {e}"),
    // → "timeout: http: error sending request..."
}

The checker statically validates that you match both Ok and Err (rule 5.3.3 — exhaustiveness on Result). If you forget a branch, the binary doesn't compile.

Canonical webhook dispatcher (the real case)

We combine the whole Fitz stack in <100 LoC. An HTTP handler receives an event, dispatches to an upstream webhook without waiting for it, responds 202 to the client immediately:

@server(8080)
fn main() => 0

let WEBHOOK_URL: Str = config("WEBHOOK_URL")

type EventInput {
    event: Str,
    user: Str,
}

@background
async fn dispatch_webhook(event: Str, user: Str) -> Null {
    let payload: Map<Str, Str> = {
        "event": event,
        "user": user,
        "source": "fitz-app",
    }
    log.info("dispatching", event: event, user: user)

    let r = http.post(WEBHOOK_URL, payload).await
    match r {
        Ok(resp) => {
            if (resp.status >= 200 and resp.status < 300) {
                log.info(
                    "delivered",
                    event: event,
                    status: resp.status,
                    duration_ms: resp.duration_ms,
                )
            } else {
                log.warn("rejected", event: event, status: resp.status, body: resp.body)
            }
        }
        Err(e) => log.error("failed", event: event, error: e),
    }
    return null
}

@post("/events")
fn create_event(input: EventInput) {
    // Fire-and-forget — handler doesn't wait for the webhook.
    let _ = spawn(dispatch_webhook(input.event, input.user))

    return 202 {
        "status": "accepted",
        "event": input.event,
        "user": input.user,
    }
}

No Celery, no RabbitMQ, no separate worker, no pip install anything. The spawn starts a fire-and-forget tokio task. The handler responds 202 in <10 ms, the upstream webhook keeps running in the background.

Quick comparison vs the same in FastAPI:

# FastAPI requires BackgroundTasks + httpx + a global client managed by hand.
from fastapi import BackgroundTasks
import httpx
import os

client = httpx.AsyncClient(timeout=30)
WEBHOOK_URL = os.environ["WEBHOOK_URL"]

async def dispatch_webhook(event: str, user: str):
    payload = {"event": event, "user": user, "source": "myapp"}
    logger.info("dispatching", extra={"event": event, "user": user})
    try:
        r = await client.post(WEBHOOK_URL, json=payload)
        if 200 <= r.status_code < 300:
            logger.info("delivered", extra={"event": event, "status": r.status_code})
        else:
            logger.warning("rejected", extra={"event": event, "status": r.status_code, "body": r.text})
    except httpx.RequestError as e:
        logger.error("failed", extra={"event": event, "error": str(e)})

@app.post("/events", status_code=202)
async def create_event(input: EventInput, bg: BackgroundTasks):
    bg.add_task(dispatch_webhook, input.event, input.user)
    return {"status": "accepted", "event": input.event, "user": input.user}

Works, but adds httpx + global client management + BackgroundTasks + structured logger setup.

fitzwatch-style health checker

The case that unblocked the mini-track. Cron job every 30 seconds pinging endpoints with http.head:

type HealthResult {
    url: Str,
    up: Bool,
    status: Int = 0,
    duration_ms: Int = 0,
    error: Str?,
}

async fn check_one(url: Str) -> HealthResult {
    let resp = http.request({
        "method": "HEAD",
        "url": url,
        "timeout_ms": 5000,
        "follow_redirects": true,
    }).await
    match resp {
        Ok(r) => {
            return HealthResult {
                url: url, up: r.status < 400,
                status: r.status, duration_ms: r.duration_ms,
            }
        }
        Err(e) => {
            return HealthResult { url: url, up: false, error: e }
        }
    }
}

@cron("*/30 * * * * *")
async fn check_all() -> Null {
    let r1 = check_one("https://example.com").await
    if (r1.up) {
        log.info("health.up", url: r1.url, status: r1.status, duration_ms: r1.duration_ms)
    } else {
        log.warn("health.down", url: r1.url, error: r1.error)
    }
    // ...repeat for more targets
    return null
}

Without @server, the binary stays alive blocking with automatic ctrl_c().await — perfect for systemd / Docker unit.

Why it matters

The 5 differentiators that justify the feature:

Language built-in — part of the fitz binary. No pip install requests / npm install axios / cargo add reqwest. When someone starts a Fitz project, they already have it.
Bit-for-bit parity fitz run ↔ fitz build — the standalone binary has reqwest statically linked with rustls-tls. Same behavior running in interpreter or production binary.
First-class async citizen — the 6 builtins return Future<Result<HttpClientResponse>>. They integrate naturally with @cron/@background/HTTP handlers/spawn(...) without extra glue.
Automatic Result<T> — transport errors as values. ? propagates, the checker requires static handling (rule 5.3.3). No silent try/catch.
No external deps on the host — rustls-tls doesn't require libssl/openssl. The binary you produce in CI runs on any target machine without system libraries.

None of the neighbor stacks in the table provides outbound HTTP client as a language built-in with this combination.

What's next

Next focus: resume fitzwatch. The technical blocker is closed, the rest is product:

src/checks.fitz with the HTTP check runner (which now actually writes the natural http.head).
src/scheduler.fitz with @cron that scans due monitors and fires spawn(run_check(m.id)).
/public/status page with server-side rendered HTML.
@ws("/dashboard") for authenticated real-time updates.

Full-stack open-source status page in pure Fitz, single-binary deploy. No separate Postgres in the 90% case (SQLite is also native + cron-driven syncing).

Try it

# Linux/Mac
curl -fsSL https://raw.githubusercontent.com/Thegreekman76/fitz/main/install.sh | sh
# Windows with scoop
scoop bucket add fitz https://github.com/Thegreekman76/scoop-fitz
scoop install fitz

# Verify
fitz --version  # → fitz 0.17.0

Runnable examples in the repo: examples/guide/17e-http-client-basico.fitz (the 5 common methods) · 17f-http-client-errores.fitz (timeout/DNS/4xx/5xx handling) · 17g-http-client-webhook.fitz (canonical dispatcher) · 17h-http-client-health-checker.fitz (cron+health checks).

Full feature doc in chapter 17 of the guide (subsection "HTTP client outbound" — Spanish).

Repo: github.com/Thegreekman76/fitz. Issues, ideas, and PRs welcome.

HTTP client outbound como ciudadano de primera clase: cero deps, paridad bit-a-bit, async natural

Martin Palopoli — Wed, 08 Jul 2026 10:04:46 +0000

Webhooks, scraping, health checks, proxying. Cada lenguaje resuelve HTTP client outbound con una librería externa. En Fitz, http.get/post/put/delete/head/request son builtins del lenguaje. Async desde día uno. Paridad bit-a-bit entre el intérprete y el binario nativo. Sin pip install requests / npm install axios / cargo add reqwest.

El detalle que se olvida

Fitz tiene @get/@post desde Fase 4. Es trivial servir HTTP. Pero hasta hace una semana, si tu handler necesitaba llamar a otro servicio (webhook a Slack, scraping, health check, proxying a un upstream), tenías dos opciones:

Importar Python con from python import urllib.request y bundlear CPython (~22 MB extra al binario).
No hacerlo.

Cualquier proyecto serio termina necesitando esto eventualmente. Detectamos el gap construyendo fitzwatch (status page open-source escrito en Fitz puro): el corazón del producto es http.head(monitor.target) cada N segundos. Sin esa línea, el lenguaje no llegaba.

Una semana después, los 6 builtins están en main, con paridad bit-a-bit entre fitz run y fitz build, validación estática vía checker, y zero deps externas en el binario final.

El stack típico de Python

pip install requests
# o pip install httpx (si querés async)

import requests

r = requests.get("https://api.example.com/users/42", timeout=5)
if r.status_code == 200:
    user = r.json()
    print(user)
elif r.status_code == 404:
    print("not found")
else:
    raise Exception(f"unexpected status: {r.status_code}")

# Async version con httpx
import httpx
async with httpx.AsyncClient() as client:
    r = await client.get("https://api.example.com/users/42", timeout=5)
    user = r.json()

Dos librerías para soportar sync y async. requests (sync, no integra con asyncio) o httpx (async, otra API).

El stack típico de JS/TS

npm install axios
# o sólo fetch del std si Node 18+

import axios from "axios"

const r = await axios.get("https://api.example.com/users/42", { timeout: 5000 })
if (r.status === 200) {
    console.log(r.data)
} else if (r.status === 404) {
    console.log("not found")
} else {
    throw new Error(`unexpected status: ${r.status}`)
}

axios rechaza con throw cualquier status >= 400 por default — querés desactivarlo o manejar el try/catch.

El stack típico de Rust

# Cargo.toml
[dependencies]
reqwest = { version = "0.12", features = ["json"] }
tokio = { version = "1", features = ["full"] }
serde_json = "1"

let client = reqwest::Client::new();
let r = client
    .get("https://api.example.com/users/42")
    .timeout(Duration::from_secs(5))
    .send()
    .await?;

match r.status().as_u16() {
    200 => {
        let user: serde_json::Value = r.json().await?;
        println!("{}", user);
    }
    404 => println!("not found"),
    other => eprintln!("unexpected status: {}", other),
}

Tres deps, dos crates a importar, un Client que construir. Funciona muy bien — reqwest es excelente — pero es un mundo aparte del std.

Lo mismo en Fitz

async fn run() -> Result<Null> {
    let r = http.get("https://api.example.com/users/42").await?
    match r.status {
        200 => print("user: {r.body}"),
        404 => print("not found"),
        s => print("unexpected status: {s}"),
    }
    return Ok(null)
}

// Top-level
match run().await {
    Ok(_) => print(""),
    Err(e) => print("transport error: {e}"),
}

Cero pip install. Cero npm install. Cero cargo add. El módulo http es parte del binario fitz.

La tabla cruda

Pieza	Python (`requests`/`httpx`)	JS (`axios`)	Rust (`reqwest`)	Fitz
Cómo se trae	`pip install`	`npm install`	`cargo add`	built-in
Sync + Async	2 libs distintas	1 lib	1 lib	1 builtin async
Error en 4xx/5xx	depende (axios sí, requests no)	sí por default	no,`Result<Response>` solo de transporte	no,`r.status`
Errores de transporte	excepción	excepción	`Result::Err`	`Result::Err(Str)`
Body JSON automático	manual `.json()`	automático en `r.data`	manual `.json::<T>().await?`	mandás `Map<Str, Any>` y se serializa
`Content-Type: application/json` automático al mandar	manual con `json=`	sí en axios	manual con `.json(&v)`	sí cuando body es `Map`
TLS	depende del OS / OpenSSL	nativo	feature flag	`rustls-tls` estático, sin OpenSSL en el host
Schema OpenAPI integrado	no aplica	no aplica	no aplica	usás los mismos `type` del server-side

Los 6 builtins, todos en una página

// GET sin body
let r = http.get("https://example.com/api").await?

// HEAD — útil para health checks (sin descargar body)
let r = http.head("https://example.com/healthz").await?

// DELETE sin body
let r = http.delete("https://example.com/api/items/42").await?

// POST con body Map (auto-JSON + Content-Type)
let r = http.post(
    "https://example.com/api/items",
    {"name": "ada", "role": "admin"}
).await?

// PUT con body Str (sin tocar headers)
let r = http.put("https://example.com/api/raw", "hello world").await?

// POST con body Bytes
let bin: Bytes = bytes([0x89, 0x50, 0x4e, 0x47])  // PNG signature
let r = http.post("https://example.com/upload", bin).await?

// request — low-level con opciones
let r = http.request({
    "method": "PATCH",
    "url": "https://example.com/api/items/42",
    "timeout_ms": 5000,
    "headers": {"X-Token": "abc"},
    "body": {"status": "active"},
    "follow_redirects": false,
}).await?

// Respuesta: 4 fields tipados
print("status      = {r.status}")
print("body        = {r.body}")
print("duration_ms = {r.duration_ms}")
let server: Str = match r.headers.get("server") {
    Ok(v) => v,
    Err(_) => "(unknown)",
}
print("server      = {server}")

r siempre es un HttpClientResponse { status: Int, body: Str, headers: Map<Str, Str>, duration_ms: Int }. El tipo es built-in del lenguaje, paralelo a Request/Response del HTTP server-side.

Manejo de errores: el delta importante

Decidimos algo distinto al stack vecino: status 4xx/5xx NO son errores. Si el server responde 404 o 500, lo recibís como Ok(response) con r.status set. Sólo los errores de transporte (DNS, timeout, TLS handshake, conexión rota) bajan a Result::Err(Str).

¿Por qué? Porque 4xx/5xx tienen body útil. Si tu API responde 422 con {"error": "validation failed", "field": "email"}, querés leer eso. Hacerlo un throw / Err pierde el body o te obliga a un try/catch que captura demasiado.

let r = http.post("https://api.example.com/users", {"email": "bad"}).await?

if (r.status >= 400 and r.status < 500) {
    // Tu API te está diciendo algo. Leelo.
    print("client error: {r.body}")
} else if (r.status >= 500) {
    print("server crashed: {r.status}")
} else {
    print("OK: {r.body}")
}

Los Err solo aparecen cuando no hubo respuesta:

match http.get("https://no-existe.example").await {
    Ok(r)  => print("status: {r.status}"),
    Err(e) => print("transport: {e}"),
    // → "transport: http: error sending request..."
}

match http.request({"method": "GET", "url": "https://slow.example", "timeout_ms": 100}).await {
    Ok(r)  => print("alcanzamos"),
    Err(e) => print("timeout: {e}"),
    // → "timeout: http: error sending request..."
}

El checker valida estáticamente que matchees Ok y Err (regla 5.3.3 — exhaustividad sobre Result). Si te olvidás un brazo, el binario no compila.

Webhook dispatcher canónico (el caso real)

Combinamos todo el stack Fitz en <100 LoC. Handler HTTP recibe un evento, dispatcha a un webhook upstream sin esperarlo, responde 202 al cliente inmediato:

@server(8080)
fn main() => 0

let WEBHOOK_URL: Str = config("WEBHOOK_URL")

type EventInput {
    event: Str,
    user: Str,
}

@background
async fn dispatch_webhook(event: Str, user: Str) -> Null {
    let payload: Map<Str, Str> = {
        "event": event,
        "user": user,
        "source": "fitz-app",
    }
    log.info("dispatching", event: event, user: user)

    let r = http.post(WEBHOOK_URL, payload).await
    match r {
        Ok(resp) => {
            if (resp.status >= 200 and resp.status < 300) {
                log.info(
                    "delivered",
                    event: event,
                    status: resp.status,
                    duration_ms: resp.duration_ms,
                )
            } else {
                log.warn("rejected", event: event, status: resp.status, body: resp.body)
            }
        }
        Err(e) => log.error("failed", event: event, error: e),
    }
    return null
}

@post("/events")
fn create_event(input: EventInput) {
    // Fire-and-forget — handler no espera al webhook.
    let _ = spawn(dispatch_webhook(input.event, input.user))

    return 202 {
        "status": "accepted",
        "event": input.event,
        "user": input.user,
    }
}

Sin Celery, sin RabbitMQ, sin worker separado, sin pip install nada. El spawn arranca un task tokio fire-and-forget. El handler responde 202 en <10 ms, el webhook upstream sigue corriendo en background.

Comparación rápida vs lo mismo en FastAPI:

# FastAPI requiere BackgroundTasks + httpx + un client global manejado a mano.
from fastapi import BackgroundTasks
import httpx
import os

client = httpx.AsyncClient(timeout=30)
WEBHOOK_URL = os.environ["WEBHOOK_URL"]

async def dispatch_webhook(event: str, user: str):
    payload = {"event": event, "user": user, "source": "myapp"}
    logger.info("dispatching", extra={"event": event, "user": user})
    try:
        r = await client.post(WEBHOOK_URL, json=payload)
        if 200 <= r.status_code < 300:
            logger.info("delivered", extra={"event": event, "status": r.status_code})
        else:
            logger.warning("rejected", extra={"event": event, "status": r.status_code, "body": r.text})
    except httpx.RequestError as e:
        logger.error("failed", extra={"event": event, "error": str(e)})

@app.post("/events", status_code=202)
async def create_event(input: EventInput, bg: BackgroundTasks):
    bg.add_task(dispatch_webhook, input.event, input.user)
    return {"status": "accepted", "event": input.event, "user": input.user}

Funciona, pero suma httpx + manejo del client global + BackgroundTasks + setup de logger estructurado.

Health checker estilo fitzwatch

El caso que destrabó la mini-tanda. Cron job cada 30 segundos que pingea endpoints con http.head:

type HealthResult {
    url: Str,
    up: Bool,
    status: Int = 0,
    duration_ms: Int = 0,
    error: Str?,
}

async fn check_one(url: Str) -> HealthResult {
    let resp = http.request({
        "method": "HEAD",
        "url": url,
        "timeout_ms": 5000,
        "follow_redirects": true,
    }).await
    match resp {
        Ok(r) => {
            return HealthResult {
                url: url, up: r.status < 400,
                status: r.status, duration_ms: r.duration_ms,
            }
        }
        Err(e) => {
            return HealthResult { url: url, up: false, error: e }
        }
    }
}

@cron("*/30 * * * * *")
async fn check_all() -> Null {
    let r1 = check_one("https://example.com").await
    if (r1.up) {
        log.info("health.up", url: r1.url, status: r1.status, duration_ms: r1.duration_ms)
    } else {
        log.warn("health.down", url: r1.url, error: r1.error)
    }
    // ...repetir para más targets
    return null
}

Sin @server, el binario queda vivo bloqueante con ctrl_c().await automático — perfecto para systemd / unit Docker.

Por qué importa

Los 5 diferenciales que justifican el feature:

Built-in del lenguaje — parte del binario fitz. Sin pip install requests / npm install axios / cargo add reqwest. Cuando alguien arranca un proyecto Fitz, ya lo tiene.
Paridad bit-a-bit fitz run ↔ fitz build — el binario standalone tiene reqwest linkeado estático con rustls-tls. Mismo comportamiento corriendo en intérprete o en binario producción.
Async ciudadano de primera — los 6 builtins devuelven Future<Result<HttpClientResponse>>. Se integran natural con @cron/@background/handlers HTTP/spawn(...) sin glue extra.
Result<T> automático — errores de transporte como valores. El ? propaga, el checker exige manejo estático (regla 5.3.3). Sin try/catch silencioso.
Sin deps externas en el host — rustls-tls no exige libssl/openssl. El binario que producís en CI corre en cualquier máquina del triple destino sin librerías del sistema.

Ninguno de los stacks vecinos del cuadro provee HTTP client outbound como builtin del lenguaje con esta combinación.

Lo que viene

Próximo norte: retomar fitzwatch. El blocker técnico está cerrado, el resto es producto:

src/checks.fitz con runner del check HTTP (que ahora sí escribe el http.head natural).
src/scheduler.fitz con @cron que escanea due monitors y dispara spawn(run_check(m.id)).
/public/status page con HTML server-side rendered.
@ws("/dashboard") para updates en tiempo real autenticados.

Status page open-source full-stack en Fitz puro, deploy de un solo binario. Sin Postgres separado en el caso 90% (SQLite también es nativo + cron-driven syncing).

Probarlo

# Linux/Mac
curl -fsSL https://raw.githubusercontent.com/Thegreekman76/fitz/main/install.sh | sh
# Windows con scoop
scoop bucket add fitz https://github.com/Thegreekman76/scoop-fitz
scoop install fitz

# Verificar
fitz --version  # → fitz 0.17.0

Ejemplos runnable en el repo: examples/guide/17e-http-client-basico.fitz (los 5 métodos comunes) · 17f-http-client-errores.fitz (manejo de timeout/DNS/4xx/5xx) · 17g-http-client-webhook.fitz (dispatcher canónico) · 17h-http-client-health-checker.fitz (cron+health checks).

Doc completo del feature en el cap 17 de la guía (sub-sección "HTTP client outbound").

Repo: github.com/Thegreekman76/fitz. Issues, ideas y PRs bienvenidos.

ORM tipado con migraciones automáticas: Fitz vs SQLAlchemy + Alembic + Pydantic

Martin Palopoli — Fri, 03 Jul 2026 11:01:22 +0000

Para tener un ORM tipado con migraciones automáticas en Python necesitás mantener 3 fuentes de verdad (SQLAlchemy + Pydantic + Alembic). En Fitz es UN type con decoradores. Plus 8× RPS y 5× menos memoria que SQLAlchemy en bench reproducible.

El doble (triple) tipado del stack Python

Para cada entidad de tu DB en una API FastAPI moderna tenés que mantener:

models.py — modelo SQLAlchemy (define el schema).
schemas.py — modelos Pydantic (define request/response).
alembic/versions/*.py — migraciones autogeneradas.

# models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String, nullable=False, unique=True)
    name = Column(String, nullable=False)
    role = Column(String, nullable=False, default="user")
    created_at = Column(DateTime, default=datetime.utcnow)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    title = Column(String, nullable=False)
    body = Column(String, nullable=False)
    user = relationship("User", back_populates="posts")

# schemas.py
from pydantic import BaseModel
from datetime import datetime

class UserCreate(BaseModel):
    email: str
    name: str
    role: str = "user"

class UserOut(BaseModel):
    id: int
    email: str
    name: str
    role: str
    created_at: datetime

    class Config:
        from_attributes = True

class PostCreate(BaseModel):
    title: str
    body: str

class PostOut(BaseModel):
    id: int
    user_id: int
    title: str
    body: str

    class Config:
        from_attributes = True

class UserWithPosts(UserOut):
    posts: list[PostOut] = []

# main.py
@app.post("/users", response_model=UserOut)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_session)):
    new_user = User(**user.model_dump())
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user

@app.get("/users/{id}", response_model=UserWithPosts)
async def get_user(id: int, db: AsyncSession = Depends(get_session)):
    result = await db.execute(
        select(User).options(selectinload(User.posts)).where(User.id == id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "user not found")
    return user

Tres archivos. Dos modelos de la misma entidad mantenidos en paralelo. Y para el schema:

$ alembic revision --autogenerate -m "add user role"
# Revisás el archivo generado — autogenerate a veces se confunde
$ alembic upgrade head

Plus configurar Alembic (env.py, alembic.ini, target_metadata...).

Lo mismo en Fitz

@table("users")
type User {
    @primary id: Int,
    email: Str,
    name: Str,
    role: Str = "user",
    created_at: Str = "now()",
    @has_many("Post", "user_id") posts: List<Post>,
}

@table("posts")
type Post {
    @primary id: Int,
    user_id: Int,
    title: Str,
    body: Str,
    @belongs_to user: User?,
}

@post("/users")
async fn create_user(db: DbConn, user: User) -> Result<User> {
    return User.insert(db, user).await
}

@get("/users/{id}")
async fn get_user(db: DbConn, id: Int) -> Result<User> {
    return User.where(fn(u) => u.id == id).preload("posts").first(db).await
}

UN type. UNA fuente de verdad. El compilador genera:

El SQL parametrizado para las queries.
El schema OpenAPI desde el mismo type.
El deserializer JSON del body desde el mismo type.
La validación de fields requeridos / nullables / defaults.

Y el schema de la DB:

$ fitz db diff
+ CREATE TABLE users (
+     id BIGSERIAL PRIMARY KEY,
+     email TEXT NOT NULL,
+     name TEXT NOT NULL,
+     role TEXT NOT NULL DEFAULT 'user',
+     created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ );
+ CREATE TABLE posts (
+     id BIGSERIAL PRIMARY KEY,
+     user_id BIGINT NOT NULL REFERENCES users(id),
+     title TEXT NOT NULL,
+     body TEXT NOT NULL
+ );

$ fitz db migrate
✓ aplicada migration 20260616_120300_initial.sql

La tabla cruda

Item	Python (SQLAlchemy + Alembic + Pydantic)	Fitz
Fuentes de verdad para una entidad	3 (models.py, schemas.py, alembic/)	1 (`@table type`)
Generar schema	`alembic revision --autogenerate -m "..."`	`fitz db diff`
Aplicar	`alembic upgrade head`	`fitz db migrate`
Rollback	`alembic downgrade -1`	`fitz db rollback`
Query DSL	`.filter(User.role == "admin")` (DSL ORM)	`.where(fn(u) => u.role == "admin")` (closure-to-SQL en compile-time)
Eager loading	`.options(selectinload(User.posts))`	`.preload("posts")`
Operadores Postgres	`User.email.ilike(...)`, `.contains(...)`, etc.	Same —`u.email.ilike(...)`, `u.tags.has(...)`
Transactions	`with session.begin():` + acordate del flush	`db.transaction(fn(tx) async { ... }).await`
Validación request body	Pydantic separado	Mismo type
Schema OpenAPI	Pydantic infer	Auto desde `@table type`
Driver	libpq vía `psycopg2`/`asyncpg`	Puro Rust (sin libpq, sin tokio-postgres)
Performance vs SQLAlchemy	baseline	8× RPS, 5× memoria menos (bench abajo)

Por partes

Closure-to-SQL en compile-time

Esto es lo que más me costó implementar y de lo que más orgulloso estoy. SQLAlchemy traduce su DSL a SQL en runtime — cada .filter(User.role == "admin") arma el AST en runtime, mide overhead, allocea, y emite SQL.

Fitz lo hace en compile-time:

let admins = User.where(fn(u) => u.role == "admin")
    .order_by(fn(u) => u.created_at, "desc")
    .limit(10)
    .all(db).await?

El compilador analiza el closure fn(u) => u.role == "admin" y emite, una sola vez al compilar el binario, el string SQL constante:

SELECT id, email, name, role, created_at FROM users
WHERE role = $1 ORDER BY created_at DESC LIMIT 10

Más el código que bindea $1 = "admin". Cero overhead runtime para construir SQL — el string ya existe en el binario como &'static str.

Comparable en performance con Diesel o sqlx (que también generan SQL en compile-time vía macros), pero con sintaxis natural del lenguaje en lugar de macros procedurales.

Operadores Postgres nativos

SQLAlchemy:

admins = session.query(User).filter(User.email.ilike("%@example.com")).all()
tagged = session.query(User).filter(User.tags.contains(["admin"])).all()
pro_users = session.query(User).filter(User.metadata["plan"].astext == "pro").all()

Fitz (los operadores son métodos sobre el field):

let admins = User.where(fn(u) => u.email.ilike("%@example.com")).all(db).await?
let tagged = User.where(fn(u) => u.tags.has("admin")).all(db).await?
let pro_users = User.where(fn(u) => u.metadata.get("plan") == "pro").all(db).await?

Operadores cubiertos:

Categoría	Métodos
String	`.like(p)`, `.ilike(p)`, `.starts_with(p)`, `.ends_with(p)`, `.contains(p)` (con escape auto de `%`/`_`)
Listas/IN	`.is_in([...])`, `.is_null()`, `.is_not_null()`
Arrays	`.has(x)` (`?`), `.contains_all(xs)` (`@>`), `.contained_in(xs)` (`<@`)
JSONB	`.has_key(k)` (`?`), `.has_all_keys(ks)` (`?&`), `.has_any_keys(ks)` (`?\

Relations + eager loading

SQLAlchemy:
{% raw %}

class User(Base):
    posts = relationship("Post", back_populates="user", lazy="select")

class Post(Base):
    user = relationship("User", back_populates="posts")

# Eager loading
users = await db.execute(
    select(User).options(selectinload(User.posts))
)

Plus configurar lazy= correctamente para no caer en el N+1 más viejo del mundo (queries lazy disparando una por user).

Fitz:

@table("users")
type User {
    @primary id: Int,
    name: Str,
    @has_many("Post", "user_id") posts: List<Post>,
}

@table("posts")
type Post {
    @primary id: Int,
    user_id: Int,
    @belongs_to user: User?,
}

// Eager loading explícito (la default es NO eager — sin sorpresas)
let users = User.preload("posts").all(db).await?

.preload("posts") emite una sola query batched con WHERE user_id IN (...) y arma los structs Fitz. La default es no eager: si no preload, el field virtual posts queda vacío y vos sabés que no fuiste a la DB.

Typo en el nombre del field:

let users = User.preload("postss").all(db).await?
//                        ^^^^^^^ error de compilación: User no tiene relation "postss"

El compilador chequea el match estáticamente. SQLAlchemy te lo dice en runtime cuando el request llega.

Transactions

SQLAlchemy:

async with db.begin():
    user = User(...)
    db.add(user)
    await db.flush()  # acordate del flush sino user.id está vacío
    post = Post(user_id=user.id, ...)
    db.add(post)

Fitz:

let result = db.transaction(fn(tx) async {
    let user = User.insert(tx, User { id: 0, name: "Ada", role: "user" }).await?
    let post = Post.insert(tx, Post { id: 0, user_id: user.id, title: "hola", body: "..." }).await?
    return Ok(post)
}).await

El bloque adentro de db.transaction(...) recibe un tx: DbConn que escribe contra la transaction. Si el closure retorna Err, rollback; si retorna Ok, commit. Si la closure paniquea, rollback automático.

Y la User.insert(...) te devuelve el row con el id BIGSERIAL ya asignado, sin "acordate del flush".

Migraciones declarativas

Cambio en el type:

 @table("users")
 type User {
     @primary id: Int,
     email: Str,
     name: Str,
     role: Str = "user",
+    avatar_url: Str?,
     created_at: Str = "now()",
 }

Diff:

$ fitz db diff
+ ALTER TABLE users ADD COLUMN avatar_url TEXT;

Aplicar:

$ fitz db migrate
✓ aplicada migration 20260616_153020_add_user_avatar_url.sql

Rollback:

$ fitz db rollback
✓ revertida migration 20260616_153020_add_user_avatar_url.sql

El compilador compara los @table types del código con el schema vivo de la DB y emite las diffs idempotentes. La migración se commitea (es un archivo .sql) para review en el PR. La fase OPS sigue siendo tu responsabilidad (timing del deploy, downtime planning para ALTER COLUMN type que reescribe la tabla, etc.) — pero el SQL ya está generado y revisado.

En Alembic, --autogenerate a veces genera SQL incorrecto (especialmente para constraints, índices compuestos, defaults complejos) y siempre se recomienda revisar el archivo manualmente. Fitz emite el diff sin tocar el SQL del usuario — vos lo ves y lo applicás.

¿Cuán rápido es? — bench reproducible

Promesas vacías son fáciles. El repo trae un bench reproducible entre dos boilerplates equivalentes:

api-postgres-fitz — Fitz + ORM nativo.
api-postgres-python — FastAPI + SQLAlchemy 2.x async.

Mismo Postgres, mismos endpoints, mismo shape de respuesta, mismo docker compose. Headline en v0.10.13 (Intel Core Ultra 7 155H, Docker 29.2.1, 30s sostenidos, concurrencia 10):

Métrica	Fitz ORM	Python + SQLAlchemy	Speedup
Memory peak	9.2 MB	51 MB	5.5× más eficiente
`GET /users` p50	4.88 ms	37.85 ms	7.76×
`GET /users` RPS	1944	246	7.91×
`GET /users/{id}` p50	3.60 ms	31.87 ms	8.85×
`GET /users/{id}` RPS	2604	296	8.80×
Cold start	0.14 s	0.22 s	1.57×
Image size	131 MB	258 MB	2× más liviano

Reproducí los números con bash benchmarks/orm-vs-sqlalchemy/run.sh (~5-8 min con cache Docker caliente; requiere oha + jq). Metodología completa, output crudo y dónde la comparación es *in*justa para Fitz (ej.: SQLAlchemy hace más trabajo del lado del ORM con tracking de identity map) están en el README del bench.

Por qué Fitz es rápido: SQL constante en compile-time (cero overhead runtime de construir queries), driver Postgres puro Rust (sin libpq overhead), serialización JSON con tipos concretos (sin reflection), Arc<Mutex<>> para shared state HTTP (multi-thread sin GIL).

Por qué SQLAlchemy es lento en este test: identity map, lazy loading flags, eventos, descriptors de columnas — muchas features que valen su costo cuando las necesitás, no cuando no.

Lo que Fitz NO te da (todavía)

Honestidad sobre las deudas:

Solo Postgres. No MySQL, no SQLite. Postgres es la elección por el shape del ecosystem (binary protocol claro, tipos ricos, JSONB, arrays). MySQL/SQLite quedan para cuando aparezca demanda.
BelongsTo en .preload(...) — sólo HasMany/HasOne en eager loading por ahora. BelongsTo eager: deuda residual.
Composite PKs — @primary solo sobre un field. Tables con PK compuesta (raro en proyectos nuevos): no soportadas.
JSON operators de Postgres en .where — los más usados están (.has_key, .contains_json, .get), pero jsonb_path_query y similares solo via db.query(...) raw.
Read replicas / sharding. La conn pool es contra una sola DB. Para read replicas hay que coordinar a mano.
Migraciones con cero downtime para columns con NOT NULL sin default. El diff lo emite pero la fase OPS sigue siendo manual.
Performance tooling (EXPLAIN ANALYZE integrado en LSP) — no en MVP.

Escape hatch: SQL crudo cuando lo necesites

Para CTEs complejas, window functions, queries que el ORM no cubre:

let result = db.query("
    WITH ranked AS (
        SELECT id, name, ROW_NUMBER() OVER (PARTITION BY role ORDER BY created_at DESC) AS rn
        FROM users
    )
    SELECT * FROM ranked WHERE rn <= 3
", []).await?

db.query(sql, params) devuelve List<Map<Str, Any>>. Sin chequeo estático del SQL ni de los column names — vos sabés. Para 90% del CRUD usás el ORM tipado; para queries chiflados, escape hatch.

Cierre

La queja más legítima contra los ORMs es "agregan una capa de abstracción que después tenés que romper para hacer queries serias". SQLAlchemy lo resuelve con un DSL muy expresivo, a costo de runtime overhead y curva de aprendizaje. Diesel/sqlx lo resuelven generando SQL en compile-time, a costo de macros procedurales y verbosidad.

Fitz toma la ruta de Diesel/sqlx pero con sintaxis del lenguaje (no macros), un type checker que valida los closure-to-SQL, y migraciones del schema integradas como sub-comando del binario. Y porque está en el lenguaje, el mismo @table type se usa como request body de HTTP, como response, como argumento de User.insert(...), como elemento de List<User> que serializa a JSON — UNA fuente de verdad.

Si tu proyecto cabe en Postgres y no necesitás composite PKs ni read replicas en el corto plazo, el ORM de Fitz cierra el ciclo "request HTTP → DB → response" con menos partes móviles y mejor performance que el stack típico Python.

Si tu proyecto necesita uno de los items de "no está en el MVP", interop Python sigue siendo una opción válida (from python import sqlalchemy), o la deuda en cuestión vive en el roadmap.

Esta cierra la serie de 9 posts: presentación, tutorial, deployment, CLI builder, WebSockets, cron jobs, auth, observability, y ORM. El stack entero. Lo que sigue son releases del lenguaje + el cap 31 de la guía (Postgres + ORM nativo) si querés profundizar.

Repo: github.com/Thegreekman76/fitz
Capítulo 31 de la guía (Postgres + ORM nativo): docs/guide.md
Doc dedicado de DB y ORM (~2600 LoC, 30 secciones): docs/db-orm.md
Bench reproducible: benchmarks/orm-vs-sqlalchemy

Typed ORM with automatic migrations: Fitz vs SQLAlchemy + Alembic + Pydantic

Martin Palopoli — Fri, 03 Jul 2026 11:01:05 +0000

For a typed ORM with automatic migrations in Python you need to maintain 3 sources of truth (SQLAlchemy + Pydantic + Alembic). In Fitz it's ONE type with decorators. Plus 8× RPS and 5× less memory than SQLAlchemy in a reproducible benchmark.

The double (triple) typing of the Python stack

For each entity of your DB in a modern FastAPI you have to maintain:

models.py — SQLAlchemy model (defines the schema).
schemas.py — Pydantic models (defines request/response).
alembic/versions/*.py — autogenerated migrations.

# models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String, nullable=False, unique=True)
    name = Column(String, nullable=False)
    role = Column(String, nullable=False, default="user")
    created_at = Column(DateTime, default=datetime.utcnow)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    title = Column(String, nullable=False)
    body = Column(String, nullable=False)
    user = relationship("User", back_populates="posts")

# schemas.py
from pydantic import BaseModel
from datetime import datetime

class UserCreate(BaseModel):
    email: str
    name: str
    role: str = "user"

class UserOut(BaseModel):
    id: int
    email: str
    name: str
    role: str
    created_at: datetime

    class Config:
        from_attributes = True

class PostCreate(BaseModel):
    title: str
    body: str

class PostOut(BaseModel):
    id: int
    user_id: int
    title: str
    body: str

    class Config:
        from_attributes = True

class UserWithPosts(UserOut):
    posts: list[PostOut] = []

# main.py
@app.post("/users", response_model=UserOut)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_session)):
    new_user = User(**user.model_dump())
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user

@app.get("/users/{id}", response_model=UserWithPosts)
async def get_user(id: int, db: AsyncSession = Depends(get_session)):
    result = await db.execute(
        select(User).options(selectinload(User.posts)).where(User.id == id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "user not found")
    return user

Three files. Two models of the same entity maintained in parallel. And for the schema:

$ alembic revision --autogenerate -m "add user role"
# Review the generated file — autogenerate sometimes gets confused
$ alembic upgrade head

Plus configuring Alembic (env.py, alembic.ini, target_metadata...).

The same thing in Fitz

@table("users")
type User {
    @primary id: Int,
    email: Str,
    name: Str,
    role: Str = "user",
    created_at: Str = "now()",
    @has_many("Post", "user_id") posts: List<Post>,
}

@table("posts")
type Post {
    @primary id: Int,
    user_id: Int,
    title: Str,
    body: Str,
    @belongs_to user: User?,
}

@post("/users")
async fn create_user(db: DbConn, user: User) -> Result<User> {
    return User.insert(db, user).await
}

@get("/users/{id}")
async fn get_user(db: DbConn, id: Int) -> Result<User> {
    return User.where(fn(u) => u.id == id).preload("posts").first(db).await
}

ONE type. ONE source of truth. The compiler generates:

The parameterized SQL for queries.
The OpenAPI schema from the same type.
The body JSON deserializer from the same type.
The validation of required / nullable / default fields.

And the DB schema:

$ fitz db diff
+ CREATE TABLE users (
+     id BIGSERIAL PRIMARY KEY,
+     email TEXT NOT NULL,
+     name TEXT NOT NULL,
+     role TEXT NOT NULL DEFAULT 'user',
+     created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ );
+ CREATE TABLE posts (
+     id BIGSERIAL PRIMARY KEY,
+     user_id BIGINT NOT NULL REFERENCES users(id),
+     title TEXT NOT NULL,
+     body TEXT NOT NULL
+ );

$ fitz db migrate
✓ applied migration 20260616_120300_initial.sql

The raw table

Item	Python (SQLAlchemy + Alembic + Pydantic)	Fitz
Sources of truth per entity	3 (models.py, schemas.py, alembic/)	1 (`@table type`)
Generate schema	`alembic revision --autogenerate -m "..."`	`fitz db diff`
Apply	`alembic upgrade head`	`fitz db migrate`
Rollback	`alembic downgrade -1`	`fitz db rollback`
Query DSL	`.filter(User.role == "admin")` (ORM DSL)	`.where(fn(u) => u.role == "admin")` (closure-to-SQL at compile time)
Eager loading	`.options(selectinload(User.posts))`	`.preload("posts")`
Postgres operators	`User.email.ilike(...)`, `.contains(...)`, etc.	Same —`u.email.ilike(...)`, `u.tags.has(...)`
Transactions	`with session.begin():` + remember to flush	`db.transaction(fn(tx) async { ... }).await`
Body request validation	Separate Pydantic	Same type
OpenAPI schema	Pydantic infer	Auto from `@table type`
Driver	libpq via `psycopg2`/`asyncpg`	Pure Rust (no libpq, no tokio-postgres)
Performance vs SQLAlchemy	baseline	8× RPS, 5× less memory (bench below)

Piece by piece

Closure-to-SQL at compile time

This is what took me the most effort and what I'm proudest of. SQLAlchemy translates its DSL to SQL at runtime — each .filter(User.role == "admin") builds the AST at runtime, takes overhead, allocates, and emits SQL.

Fitz does it at compile time:

let admins = User.where(fn(u) => u.role == "admin")
    .order_by(fn(u) => u.created_at, "desc")
    .limit(10)
    .all(db).await?

The compiler analyzes the closure fn(u) => u.role == "admin" and emits, once at binary compile time, the constant SQL string:

SELECT id, email, name, role, created_at FROM users
WHERE role = $1 ORDER BY created_at DESC LIMIT 10

Plus the code that binds $1 = "admin". Zero runtime overhead for SQL construction — the string already exists in the binary as &'static str.

Comparable in performance to Diesel or sqlx (which also generate SQL at compile time via macros), but with natural language syntax instead of procedural macros.

Native Postgres operators

SQLAlchemy:

admins = session.query(User).filter(User.email.ilike("%@example.com")).all()
tagged = session.query(User).filter(User.tags.contains(["admin"])).all()
pro_users = session.query(User).filter(User.metadata["plan"].astext == "pro").all()

Fitz (operators are methods on the field):

let admins = User.where(fn(u) => u.email.ilike("%@example.com")).all(db).await?
let tagged = User.where(fn(u) => u.tags.has("admin")).all(db).await?
let pro_users = User.where(fn(u) => u.metadata.get("plan") == "pro").all(db).await?

Operators covered:

Category	Methods
String	`.like(p)`, `.ilike(p)`, `.starts_with(p)`, `.ends_with(p)`, `.contains(p)` (with auto-escape of `%`/`_`)
Lists/IN	`.is_in([...])`, `.is_null()`, `.is_not_null()`
Arrays	`.has(x)` (`?`), `.contains_all(xs)` (`@>`), `.contained_in(xs)` (`<@`)
JSONB	`.has_key(k)` (`?`), `.has_all_keys(ks)` (`?&`), `.has_any_keys(ks)` (`?\

Relations + eager loading

SQLAlchemy:
{% raw %}

class User(Base):
    posts = relationship("Post", back_populates="user", lazy="select")

class Post(Base):
    user = relationship("User", back_populates="posts")

# Eager loading
users = await db.execute(
    select(User).options(selectinload(User.posts))
)

Plus configuring lazy= correctly to avoid the oldest N+1 in the world (lazy queries firing one per user).

Fitz:

@table("users")
type User {
    @primary id: Int,
    name: Str,
    @has_many("Post", "user_id") posts: List<Post>,
}

@table("posts")
type Post {
    @primary id: Int,
    user_id: Int,
    @belongs_to user: User?,
}

// Explicit eager loading (default is NOT eager — no surprises)
let users = User.preload("posts").all(db).await?

.preload("posts") emits a single batched query with WHERE user_id IN (...) and builds the Fitz structs. The default is non-eager: if you don't preload, the virtual field posts is empty and you know you didn't hit the DB.

Typo in the field name:

let users = User.preload("postss").all(db).await?
//                        ^^^^^^^ compile error: User has no relation "postss"

The compiler statically checks the match. SQLAlchemy tells you at runtime when the request lands.

Transactions

SQLAlchemy:

async with db.begin():
    user = User(...)
    db.add(user)
    await db.flush()  # remember to flush or user.id is empty
    post = Post(user_id=user.id, ...)
    db.add(post)

Fitz:

let result = db.transaction(fn(tx) async {
    let user = User.insert(tx, User { id: 0, name: "Ada", role: "user" }).await?
    let post = Post.insert(tx, Post { id: 0, user_id: user.id, title: "hi", body: "..." }).await?
    return Ok(post)
}).await

The block inside db.transaction(...) receives a tx: DbConn that writes against the transaction. If the closure returns Err, rollback; if it returns Ok, commit. If the closure panics, automatic rollback.

And User.insert(...) returns the row with the BIGSERIAL id already assigned, no "remember to flush".

Declarative migrations

Change to the type:

 @table("users")
 type User {
     @primary id: Int,
     email: Str,
     name: Str,
     role: Str = "user",
+    avatar_url: Str?,
     created_at: Str = "now()",
 }

Diff:

$ fitz db diff
+ ALTER TABLE users ADD COLUMN avatar_url TEXT;

Apply:

$ fitz db migrate
✓ applied migration 20260616_153020_add_user_avatar_url.sql

Rollback:

$ fitz db rollback
✓ reverted migration 20260616_153020_add_user_avatar_url.sql

The compiler compares the @table types in your code with the live schema and emits idempotent diffs. The migration is committed (it's a .sql file) for review in the PR. The OPS phase is still your responsibility (deploy timing, downtime planning for ALTER COLUMN type that rewrites the table, etc.) — but the SQL is already generated and reviewed.

In Alembic, --autogenerate sometimes generates incorrect SQL (especially for constraints, composite indexes, complex defaults) and you're always told to review the file manually. Fitz emits the diff without touching the user's SQL — you see it and you apply it.

How fast is it? — reproducible benchmark

Empty promises are easy. The repo ships a reproducible benchmark between two equivalent boilerplates:

api-postgres-fitz — Fitz + native ORM.
api-postgres-python — FastAPI + SQLAlchemy 2.x async.

Same Postgres, same endpoints, same response shape, same docker compose. Headline at v0.10.13 (Intel Core Ultra 7 155H, Docker 29.2.1, 30s sustained, concurrency 10):

Metric	Fitz ORM	Python + SQLAlchemy	Speedup
Memory peak	9.2 MB	51 MB	5.5× more efficient
`GET /users` p50	4.88 ms	37.85 ms	7.76×
`GET /users` RPS	1944	246	7.91×
`GET /users/{id}` p50	3.60 ms	31.87 ms	8.85×
`GET /users/{id}` RPS	2604	296	8.80×
Cold start	0.14 s	0.22 s	1.57×
Image size	131 MB	258 MB	2× lighter

Reproduce the numbers with bash benchmarks/orm-vs-sqlalchemy/run.sh (~5-8 min with warm Docker cache; requires oha + jq). Full methodology, raw output, and where the comparison is *un*fair to Fitz (e.g.: SQLAlchemy does more work on the ORM side with identity map tracking) are in the bench README.

Why Fitz is fast: constant SQL at compile time (zero runtime overhead for building queries), pure Rust Postgres driver (no libpq overhead), JSON serialization with concrete types (no reflection), Arc<Mutex<>> for shared HTTP state (multi-thread without GIL).

Why SQLAlchemy is slow in this test: identity map, lazy loading flags, events, column descriptors — many features that earn their cost when you need them, not when you don't.

What Fitz does NOT give you (yet)

Honesty about the debts:

Postgres only. No MySQL, no SQLite. Postgres is the choice for the ecosystem shape (clean binary protocol, rich types, JSONB, arrays). MySQL/SQLite are for when demand appears.
BelongsTo in .preload(...) — only HasMany/HasOne in eager loading for now. BelongsTo eager: residual debt.
Composite PKs — @primary only on one field. Tables with composite PKs (rare in new projects): not supported.
Postgres JSON operators in .where — the most used ones are there (.has_key, .contains_json, .get), but jsonb_path_query and similar only via raw db.query(...).
Read replicas / sharding. The conn pool is against a single DB. For read replicas you have to coordinate by hand.
Zero-downtime migrations for NOT NULL columns without default. The diff emits it but the OPS phase remains manual.
Performance tooling (EXPLAIN ANALYZE integrated in LSP) — not in MVP.

Escape hatch: raw SQL when you need it

For complex CTEs, window functions, queries the ORM doesn't cover:

let result = db.query("
    WITH ranked AS (
        SELECT id, name, ROW_NUMBER() OVER (PARTITION BY role ORDER BY created_at DESC) AS rn
        FROM users
    )
    SELECT * FROM ranked WHERE rn <= 3
", []).await?

db.query(sql, params) returns List<Map<Str, Any>>. No static checking of SQL or column names — you know. For 90% of CRUD you use the typed ORM; for crazy queries, escape hatch.

Closing

The most legitimate complaint against ORMs is "they add a layer of abstraction that you then have to break to do serious queries". SQLAlchemy solves this with a highly expressive DSL, at the cost of runtime overhead and learning curve. Diesel/sqlx solve it by generating SQL at compile time, at the cost of procedural macros and verbosity.

Fitz takes the Diesel/sqlx route but with language syntax (not macros), a type checker that validates the closure-to-SQL, and schema migrations integrated as a binary subcommand. And because it's in the language, the same @table type is used as an HTTP request body, as a response, as an argument to User.insert(...), as an element of List<User> that serializes to JSON — ONE source of truth.

If your project fits in Postgres and you don't need composite PKs or read replicas in the short term, Fitz's ORM closes the cycle "HTTP request → DB → response" with fewer moving parts and better performance than the typical Python stack.

If your project needs one of the "not in MVP" items, Python interop is still a valid option (from python import sqlalchemy), or the debt in question lives in the roadmap.

This closes the series of 9 posts: intro, tutorial, deployment, CLI builder, WebSockets, cron jobs, auth, observability, and ORM. The whole stack. What's next are language releases + chapter 31 of the guide (Postgres + native ORM) if you want to go deeper.

Repo: github.com/Thegreekman76/fitz
Chapter 31 of the guide (Postgres + native ORM): docs/guide.md
Dedicated DB and ORM doc (~2600 LoC, 30 sections): docs/db-orm.md
Reproducible bench: benchmarks/orm-vs-sqlalchemy

Tracing, métricas Prometheus y logs estructurados con dos decoradores: Fitz vs el setup de OpenTelemetry en FastAPI

Martin Palopoli — Tue, 30 Jun 2026 10:28:54 +0000

Para tener observability completa en FastAPI necesitás 6 paquetes pip + 60 líneas de config + glue manual entre logs/spans/métricas. En Fitz son dos decoradores y un env var. Con trace_id correlacionado auto entre logs y spans, y Secret<T> redactado en logs sin pensar.

El stack que toda app "production-ready" termina pegoteando

Tu app crece. El cliente quiere saber qué endpoint está lento, cuántos requests fallaron en la última hora, y por qué un user específico vio un error a las 3 AM. Hablamos del Triángulo Sagrado de observability: traces, metrics, logs. En 2026 la respuesta de la industria es OpenTelemetry para los tres.

En Python con FastAPI:

pip install opentelemetry-distro[otlp] \
            opentelemetry-instrumentation-fastapi \
            opentelemetry-instrumentation-sqlalchemy \
            opentelemetry-instrumentation-requests \
            opentelemetry-exporter-otlp-proto-grpc \
            prometheus-fastapi-instrumentator \
            structlog

observability.py (~60 líneas):

import os
import logging
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
import structlog
from prometheus_fastapi_instrumentator import Instrumentator

SERVICE_NAME = os.environ.get("OTEL_SERVICE_NAME", "myapp")
OTLP_ENDPOINT = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
SAMPLE_RATIO = float(os.environ.get("OTEL_TRACES_SAMPLER_ARG", "1.0"))

def setup_observability(app, engine):
    resource = Resource.create({"service.name": SERVICE_NAME})

    if OTLP_ENDPOINT:
        trace_provider = TracerProvider(
            resource=resource,
            sampler=TraceIdRatioBased(SAMPLE_RATIO),
        )
        trace_provider.add_span_processor(
            BatchSpanProcessor(OTLPSpanExporter(endpoint=OTLP_ENDPOINT))
        )
        trace.set_tracer_provider(trace_provider)

        metric_reader = PeriodicExportingMetricReader(
            OTLPMetricExporter(endpoint=OTLP_ENDPOINT)
        )
        meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
        metrics.set_meter_provider(meter_provider)

    FastAPIInstrumentor.instrument_app(app)
    SQLAlchemyInstrumentor().instrument(engine=engine)
    Instrumentator().instrument(app).expose(app, endpoint="/metrics")

    # Logs estructurados con trace_id
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.dict_tracebacks,
            inject_trace_context,  # propio, ver abajo
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        cache_logger_on_first_use=True,
    )

def inject_trace_context(logger, method_name, event_dict):
    span = trace.get_current_span()
    if span and span.get_span_context().is_valid:
        ctx = span.get_span_context()
        event_dict["trace_id"] = format(ctx.trace_id, "032x")
        event_dict["span_id"] = format(ctx.span_id, "016x")
    return event_dict

Plus uso en handlers:

import structlog
from opentelemetry import trace, metrics

log = structlog.get_logger()
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

orders_counter = meter.create_counter("orders_calls_total")
orders_histogram = meter.create_histogram(
    "orders_duration_seconds",
    unit="s",
)

@app.post("/orders")
async def process_order(body: OrderIn):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", body.id)
        start = time.time()
        try:
            log.info("order.processing", order_id=body.id, total=body.total)
            receipt = await actually_process(body)
            orders_counter.add(1, {"status": "success"})
            log.info("order.processed", receipt_id=receipt.id)
            return receipt
        except Exception as e:
            orders_counter.add(1, {"status": "error"})
            log.error("order.failed", order_id=body.id, error=str(e))
            raise
        finally:
            orders_histogram.record(time.time() - start)

Ocho instalaciones de paquetes. ~60 líneas de setup. ~25 líneas por handler para tracear + metricar + loguear. Conexión manual entre las tres signals.

Y ojo con: si te olvidás de llamar FastAPIInstrumentor.instrument_app(app), no hay spans HTTP. Si te olvidás del SQLAlchemyInstrumentor, no se ve la DB. Si structlog no tiene el processor inject_trace_context, los logs no se correlacionan con los spans.

Lo mismo en Fitz

@server(8080, prometheus=true)
fn main() => 0

@trace(name="process_order")
@metric(name="orders")
async fn process_order(body: OrderIn) -> Result<Receipt> {
    log.info("order.processing", { order_id: body.id, total: body.total })
    let receipt = actually_process(body).await?
    log.info("order.processed", { receipt_id: receipt.id })
    return Ok(receipt)
}

@post("/orders")
async fn create_order(body: OrderIn) -> Result<Receipt> {
    return process_order(body).await
}

Activar el export OTLP:

export OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318
fitz run main.fitz

Eso es todo.

La tabla cruda

Item	Python (OTel + structlog + 6 libs)	Fitz
Setup inicial	~60 LoC + 6 instalaciones pip	`@server(prometheus=true)`
Span HTTP por request	`FastAPIInstrumentor.instrument_app(app)`	Auto-instrumented
Span custom sobre fn	`with tracer.start_as_current_span("X")`	`@trace(name="X")`
Counter de calls	`meter.create_counter(...) + .add(1)`	`@metric(name="X")` (incluye duration histogram)
Histogram de duration	`meter.create_histogram(...) + .record(...)` + try/finally	Mismo `@metric` (RAII guard)
Endpoint `/metrics` Prometheus	`Instrumentator().instrument(app).expose(app)`	`@server(prometheus=true)`
Logs estructurados JSON	`structlog.configure(...)` con 6 processors	`log.info("event", { ... })` built-in
Trace_id propagado a logs	Processor custom `inject_trace_context`	Automático (task-local)
Secret redactado en logs	`.get_secret_value()` manual con cuidado	`Secret<T>` se redacta auto
HTTP access log	`FastAPIInstrumentor` lo emite	Auto-emitido con `trace_id`/`span_id`
Sampling tail-based	TraceIdRatioBased	TraceIdRatioBased (mismo)

Por partes

Spans HTTP auto-instrumentados

En Fitz, cada @get/@post/@put/@delete/@ws abre un span OTel con:

http.method (GET/POST/...)
http.target (el path template — /users/{id} no /users/42, low cardinality friendly)
http.status_code al cerrar el span
duration_ms

Y emite un access log con los mismos campos + trace_id/span_id. Sin código del user.

En Python tenés que llamar FastAPIInstrumentor.instrument_app(app) y rezar que tu versión matchea. Si tu user agent emite headers con caracteres no-ASCII, la version vieja de la lib panickeaba — bug famoso.

Trace_id propagado a logs custom

Adentro del span del request:

@authenticated @post("/orders")
async fn create_order(user: User, body: OrderIn) -> Result<Receipt> {
    log.info("order.received", { order_id: body.id, user_email: user.email })
    let receipt = process(body).await?
    log.info("order.ack", { receipt_id: receipt.id })
    return Ok(receipt)
}

Cada log.info(...) adentro del handler incluye automáticamente:

{
  "timestamp": "2026-06-16T10:23:01.231Z",
  "level": "info",
  "msg": "order.received",
  "trace_id": "5e4f9b2c8a7d3e1f0b6c9a4d8e2f1a3b",
  "span_id": "a1b2c3d4e5f6a7b8",
  "order_id": 42,
  "user_email": "ada@example.com"
}

El trace_id matchea con el trace_id del span en Jaeger/Tempo. Por la cierre 9.x.4 iter2.a, cuando hay OTel activo, el trace_id de los logs es exactamente el mismo del span OTel — habilita queries cross-pipeline ("dame todos los logs cuyo trace_id coincide con este span de Jaeger").

En Python tenés que escribir el processor inject_trace_context a mano (~10 líneas), agregarlo a la config de structlog, y validar que cada handler usa structlog en lugar del logging stdlib (porque si alguien hace import logging; logging.info(...) directamente, los logs NO van a tener el trace_id).

Métricas con un decorador

@metric(name="orders") automáticamente registra DOS metrics:

orders_calls_total — Counter, incrementado al return de la fn.
orders_duration_seconds — Histogram, registrado al return (incluso si la fn paniquea, por RAII guard).

@trace(name="process_order")
@metric(name="orders")
async fn process(order: Order) -> Result<Receipt> {
    // process_order span se cierra al return
    // orders_calls_total += 1
    // orders_duration_seconds.observe(elapsed)
}

En Python con OTel, para el mismo efecto:

orders_counter = meter.create_counter("orders_calls_total")
orders_histogram = meter.create_histogram("orders_duration_seconds", unit="s")

@tracer.start_as_current_span("process_order")
async def process(order: Order) -> Receipt:
    start = time.time()
    try:
        result = await actually_process(order)
        orders_counter.add(1)
        return result
    finally:
        orders_histogram.record(time.time() - start)

5× más código. Y si te olvidás el finally, la histogram pierde casos. Decoradores de Fitz garantizan el cleanup vía RAII.

Endpoint Prometheus opcional

@server(8080, prometheus=true)
fn main() => 0

Auto-monta GET /metrics en el mismo puerto, devolviendo el formato exposition de Prometheus. Sin librería separada. Sin definir el endpoint a mano.

Si el user declaró su propio @get("/metrics"), gana — mismo patrón que /openapi.json//healthz.

En Python: Instrumentator().instrument(app).expose(app) — está bien, pero es otra dep, otra responsabilidad, otra versión a matchear con FastAPI.

Export OTLP con un env var

# Para Jaeger
export OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318

# Para Honeycomb (con headers)
export OTEL_EXPORTER_OTLP_ENDPOINT=https://api.honeycomb.io
export OTEL_EXPORTER_OTLP_HEADERS=x-honeycomb-team=abc123

# Para Tempo
export OTEL_EXPORTER_OTLP_ENDPOINT=http://tempo:4318

# Tail-based sampling 10%
export OTEL_TRACES_SAMPLER_ARG=0.1

# Service name
export OTEL_SERVICE_NAME=myapp

Sin la env var, cero overhead, cero conexiones de red. El instrumento corre como no-op si no hay endpoint declarado.

Estos env vars son los standard de OpenTelemetry, no inventados por Fitz. Compatible con cualquier backend OTel (Jaeger, Tempo, Honeycomb, Datadog, NewRelic, Grafana Cloud, etc.).

`Secret<T>` redactado auto en logs

Esta es mi feature favorita. En Python:

log.info("auth.success", token=user.access_token)  # ¡bug! va el token a Loki

El bug que más vi en código de producción: alguien loguea la API key, la password, el JWT. El secret va a Loki/Sentry/Datadog. Borrar logs de producción es una operación lenta y dolorosa.

En Fitz, Secret<T> redacta automáticamente en Display y en serialización JSON:

let JWT_SECRET: Secret<Str> = secret("JWT_SECRET")
let user_token: Secret<Str> = generate_token(user)

log.info("auth.success", { token: user_token })
// → {"msg": "auth.success", "token": "***"}

print("JWT_SECRET = {JWT_SECRET}")
// → "JWT_SECRET = ***"

Para exponer el valor real (al firmar JWT, al pegar a la DB):

let token = jwt.encode(claims, JWT_SECRET.expose())

.expose() es explícito, grepeable. El code review puede auditar cada call site en segundos.

En Pydantic existe SecretStr pero tenés que recordar opt-in en cada lugar, y la gente se olvida. Fitz hace que la versión segura sea la default.

Decisiones de diseño que vale la pena entender

`@trace`/`@metric` solo sobre fns user, no sobre HTTP/WS

Los handlers HTTP y WebSocket ya tienen auto-instrumentation con el span del request. Stackear @trace arriba de @get sería redundante y crearía spans anidados sin valor. El checker lo rechaza con mensaje claro.

Acceso log auto-emitido

Cada handler HTTP emite un log.info("http.access", ...) al return con http.method/http.target/http.status_code/duration_ms. Sin opt-in. Si querés desactivarlo:

@server(8080, observability=false)
fn main() => 0

Y el wrapper de instrumentación se bypassea entero.

Storage del span context con task-local

SpanContext vive en un tokio::task_local!. Atraviesa thread boundaries en runtime multi-thread. Sin globales mutables, sin race conditions.

Lo que Fitz NO te da (todavía)

Honestidad sobre las deudas residuales de Fase 12.3 documentadas explícitamente:

Bridge logs OTel: los log.X(...) van a stderr (con trace_id propagado). Para que ALSO vayan al log signal de OTel del backend (correlacionado con spans ahí), tenés que esperar al sub-paso 12.3.iter2.b — diseñado pero no shipped.
Bridge métricas OTel: las metrics que emite @metric despachan al recorder Prometheus cuando @server(prometheus=true). Para que también vayan al OTel metrics signal (push a Honeycomb metrics, NewRelic metrics) hay que esperar release del crate upstream metrics-exporter-opentelemetry compatible con opentelemetry_sdk 0.32 (deuda documentada, no por desidia nuestra).
Sampling tail-based: solo head-based (TraceIdRatioBased). Para "exportá traces que tuvieron error" o "samplealos por latencia" hay que correr OTel collector en el medio con la config.
Profiling continuo (pyroscope/pprof) — no integrado.
Auto-instrumentation de DB: el ORM nativo emite el SQL ejecutado en los spans del handler que lo llama. Para queries vía db.query(...) raw, hoy no se crea span hijo (deuda menor).

Cierre

Observability en Python es el área donde más visiblemente paga "ser library-first": cada signal vive en una lib distinta, cada lib tiene su propia config, cada handler necesita boilerplate para usarlas, y mantener consistencia entre las tres signals es responsabilidad tuya.

Fitz mete las tres signals en el lenguaje, con auto-instrumentation para lo que toda app necesita (HTTP spans + access logs + metrics), decoradores opcionales para lo custom (@trace/@metric), y conexión nativa entre signals (trace_id propagado, Secret redactado).

El target sigue siendo el ecosistema OTel — exportás a los mismos backends, con los mismos env vars estándar. Lo que cambia es el código tuyo.

Si tu pipeline de observability ya está estabilizado con OTel en FastAPI y funciona, no hay urgencia. Si estás arrancando un proyecto en 2026 y querés evitar el día de plumbing, vale la prueba.

Próximo post de la serie: "ORM tipado con migraciones automáticas: Fitz vs SQLAlchemy + Alembic + Pydantic" — la última pieza del stack, con benchmarks reproducibles.

Repo: github.com/Thegreekman76/fitz
Capítulo 33 de la guía (Observability): docs/guide.md

Tracing, Prometheus metrics, and structured logs with two decorators: Fitz vs the OpenTelemetry setup in FastAPI

Martin Palopoli — Tue, 30 Jun 2026 10:28:39 +0000

For full observability in FastAPI you need 6 pip packages + 60 lines of config + manual glue between logs/spans/metrics. In Fitz it's two decorators and an env var. With trace_id auto-correlated between logs and spans, and Secret<T> redacted in logs without thinking.

The stack every "production-ready" app ends up gluing together

Your app grows. The client wants to know which endpoint is slow, how many requests failed in the last hour, and why a specific user saw an error at 3 AM. We're talking about the Sacred Triangle of observability: traces, metrics, logs. In 2026 the industry answer is OpenTelemetry for all three.

In Python with FastAPI:

pip install opentelemetry-distro[otlp] \
            opentelemetry-instrumentation-fastapi \
            opentelemetry-instrumentation-sqlalchemy \
            opentelemetry-instrumentation-requests \
            opentelemetry-exporter-otlp-proto-grpc \
            prometheus-fastapi-instrumentator \
            structlog

observability.py (~60 lines):

import os
import logging
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
import structlog
from prometheus_fastapi_instrumentator import Instrumentator

SERVICE_NAME = os.environ.get("OTEL_SERVICE_NAME", "myapp")
OTLP_ENDPOINT = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
SAMPLE_RATIO = float(os.environ.get("OTEL_TRACES_SAMPLER_ARG", "1.0"))

def setup_observability(app, engine):
    resource = Resource.create({"service.name": SERVICE_NAME})

    if OTLP_ENDPOINT:
        trace_provider = TracerProvider(
            resource=resource,
            sampler=TraceIdRatioBased(SAMPLE_RATIO),
        )
        trace_provider.add_span_processor(
            BatchSpanProcessor(OTLPSpanExporter(endpoint=OTLP_ENDPOINT))
        )
        trace.set_tracer_provider(trace_provider)

        metric_reader = PeriodicExportingMetricReader(
            OTLPMetricExporter(endpoint=OTLP_ENDPOINT)
        )
        meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
        metrics.set_meter_provider(meter_provider)

    FastAPIInstrumentor.instrument_app(app)
    SQLAlchemyInstrumentor().instrument(engine=engine)
    Instrumentator().instrument(app).expose(app, endpoint="/metrics")

    # Structured logs with trace_id
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.dict_tracebacks,
            inject_trace_context,  # custom, see below
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        cache_logger_on_first_use=True,
    )

def inject_trace_context(logger, method_name, event_dict):
    span = trace.get_current_span()
    if span and span.get_span_context().is_valid:
        ctx = span.get_span_context()
        event_dict["trace_id"] = format(ctx.trace_id, "032x")
        event_dict["span_id"] = format(ctx.span_id, "016x")
    return event_dict

Plus usage in handlers:

import structlog
from opentelemetry import trace, metrics

log = structlog.get_logger()
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

orders_counter = meter.create_counter("orders_calls_total")
orders_histogram = meter.create_histogram(
    "orders_duration_seconds",
    unit="s",
)

@app.post("/orders")
async def process_order(body: OrderIn):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", body.id)
        start = time.time()
        try:
            log.info("order.processing", order_id=body.id, total=body.total)
            receipt = await actually_process(body)
            orders_counter.add(1, {"status": "success"})
            log.info("order.processed", receipt_id=receipt.id)
            return receipt
        except Exception as e:
            orders_counter.add(1, {"status": "error"})
            log.error("order.failed", order_id=body.id, error=str(e))
            raise
        finally:
            orders_histogram.record(time.time() - start)

Eight pip installs. ~60 lines of setup. ~25 lines per handler to trace + meter + log. Manual connection between the three signals.

And watch out: if you forget to call FastAPIInstrumentor.instrument_app(app), no HTTP spans. If you forget SQLAlchemyInstrumentor, the DB doesn't show. If structlog doesn't have the inject_trace_context processor, logs won't correlate with spans.

The same thing in Fitz

@server(8080, prometheus=true)
fn main() => 0

@trace(name="process_order")
@metric(name="orders")
async fn process_order(body: OrderIn) -> Result<Receipt> {
    log.info("order.processing", { order_id: body.id, total: body.total })
    let receipt = actually_process(body).await?
    log.info("order.processed", { receipt_id: receipt.id })
    return Ok(receipt)
}

@post("/orders")
async fn create_order(body: OrderIn) -> Result<Receipt> {
    return process_order(body).await
}

Activate OTLP export:

export OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318
fitz run main.fitz

That's it.

The raw table

Item	Python (OTel + structlog + 6 libs)	Fitz
Initial setup	~60 LoC + 6 pip installs	`@server(prometheus=true)`
HTTP span per request	`FastAPIInstrumentor.instrument_app(app)`	Auto-instrumented
Custom span over fn	`with tracer.start_as_current_span("X")`	`@trace(name="X")`
Call counter	`meter.create_counter(...) + .add(1)`	`@metric(name="X")` (includes duration histogram)
Duration histogram	`meter.create_histogram(...) + .record(...)` + try/finally	Same `@metric` (RAII guard)
Prometheus `/metrics` endpoint	`Instrumentator().instrument(app).expose(app)`	`@server(prometheus=true)`
Structured JSON logs	`structlog.configure(...)` with 6 processors	`log.info("event", { ... })` built-in
Trace_id propagated to logs	Custom `inject_trace_context` processor	Automatic (task-local)
Secret redacted in logs	Manual `.get_secret_value()` with care	`Secret<T>` redacts auto
HTTP access log	`FastAPIInstrumentor` emits it	Auto-emitted with `trace_id`/`span_id`
Tail-based sampling	TraceIdRatioBased	TraceIdRatioBased (same)

Piece by piece

Auto-instrumented HTTP spans

In Fitz, every @get/@post/@put/@delete/@ws opens an OTel span with:

http.method (GET/POST/...)
http.target (the path template — /users/{id} not /users/42, low-cardinality friendly)
http.status_code when the span closes
duration_ms

And it emits an access log with the same fields + trace_id/span_id. Zero user code.

In Python you have to call FastAPIInstrumentor.instrument_app(app) and hope your version matches. If your user agent emits headers with non-ASCII characters, the old version of the lib used to panic — famous bug.

Trace_id propagated to custom logs

Inside the request's span:

@authenticated @post("/orders")
async fn create_order(user: User, body: OrderIn) -> Result<Receipt> {
    log.info("order.received", { order_id: body.id, user_email: user.email })
    let receipt = process(body).await?
    log.info("order.ack", { receipt_id: receipt.id })
    return Ok(receipt)
}

Every log.info(...) inside the handler automatically includes:

{
  "timestamp": "2026-06-16T10:23:01.231Z",
  "level": "info",
  "msg": "order.received",
  "trace_id": "5e4f9b2c8a7d3e1f0b6c9a4d8e2f1a3b",
  "span_id": "a1b2c3d4e5f6a7b8",
  "order_id": 42,
  "user_email": "ada@example.com"
}

The trace_id matches the trace_id of the span in Jaeger/Tempo. By the 9.x.4 iter2.a close, when OTel is active, the trace_id in the logs is the exact same one as the OTel span — enabling cross-pipeline queries ("give me all logs whose trace_id matches this Jaeger span").

In Python you have to write the inject_trace_context processor by hand (~10 lines), add it to structlog's config, and validate that every handler uses structlog instead of the logging stdlib (because if anyone does import logging; logging.info(...) directly, those logs will NOT have the trace_id).

Metrics with one decorator

@metric(name="orders") automatically registers TWO metrics:

orders_calls_total — Counter, incremented at fn return.
orders_duration_seconds — Histogram, recorded at return (even if the fn panics, via RAII guard).

@trace(name="process_order")
@metric(name="orders")
async fn process(order: Order) -> Result<Receipt> {
    // process_order span closes at return
    // orders_calls_total += 1
    // orders_duration_seconds.observe(elapsed)
}

In Python with OTel, for the same effect:

orders_counter = meter.create_counter("orders_calls_total")
orders_histogram = meter.create_histogram("orders_duration_seconds", unit="s")

@tracer.start_as_current_span("process_order")
async def process(order: Order) -> Receipt:
    start = time.time()
    try:
        result = await actually_process(order)
        orders_counter.add(1)
        return result
    finally:
        orders_histogram.record(time.time() - start)

5× more code. And if you forget the finally, the histogram loses cases. Fitz's decorators guarantee cleanup via RAII.

Optional Prometheus endpoint

@server(8080, prometheus=true)
fn main() => 0

Auto-mounts GET /metrics on the same port, returning the Prometheus exposition format. No separate library. No defining the endpoint by hand.

If the user declared their own @get("/metrics"), theirs wins — same pattern as /openapi.json//healthz.

In Python: Instrumentator().instrument(app).expose(app) — it's fine, but it's another dep, another responsibility, another version to match with FastAPI.

OTLP export with one env var

# For Jaeger
export OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318

# For Honeycomb (with headers)
export OTEL_EXPORTER_OTLP_ENDPOINT=https://api.honeycomb.io
export OTEL_EXPORTER_OTLP_HEADERS=x-honeycomb-team=abc123

# For Tempo
export OTEL_EXPORTER_OTLP_ENDPOINT=http://tempo:4318

# Tail-based sampling 10%
export OTEL_TRACES_SAMPLER_ARG=0.1

# Service name
export OTEL_SERVICE_NAME=myapp

Without the env var, zero overhead, zero network calls. The instrument runs as a no-op if no endpoint is declared.

These env vars are the OpenTelemetry standard, not invented by Fitz. Compatible with any OTel backend (Jaeger, Tempo, Honeycomb, Datadog, NewRelic, Grafana Cloud, etc.).

`Secret<T>` auto-redacted in logs

This is my favorite feature. In Python:

log.info("auth.success", token=user.access_token)  # bug! token goes to Loki

The bug I've seen most in production code: someone logs the API key, the password, the JWT. The secret goes to Loki/Sentry/Datadog. Deleting production logs is a slow and painful operation.

In Fitz, Secret<T> redacts automatically in Display and JSON serialization:

let JWT_SECRET: Secret<Str> = secret("JWT_SECRET")
let user_token: Secret<Str> = generate_token(user)

log.info("auth.success", { token: user_token })
// → {"msg": "auth.success", "token": "***"}

print("JWT_SECRET = {JWT_SECRET}")
// → "JWT_SECRET = ***"

To expose the real value (when signing JWT, when hitting the DB):

let token = jwt.encode(claims, JWT_SECRET.expose())

.expose() is explicit, greppable. Code review can audit each call site in seconds.

In Pydantic there's SecretStr but you have to remember to opt in at every place, and people forget. Fitz makes the safe version the default.

Design decisions worth understanding

`@trace`/`@metric` only on user fns, not on HTTP/WS

HTTP and WebSocket handlers already have auto-instrumentation with the request span. Stacking @trace on top of @get would be redundant and would create nested spans with no value. The checker rejects it with a clear message.

Auto-emitted access log

Every HTTP handler emits a log.info("http.access", ...) on return with http.method/http.target/http.status_code/duration_ms. No opt-in. If you want to disable it:

@server(8080, observability=false)
fn main() => 0

And the instrumentation wrapper is bypassed entirely.

Span context storage with task-local

SpanContext lives in a tokio::task_local!. Crosses thread boundaries in multi-thread runtime. No mutable globals, no race conditions.

What Fitz does NOT give you (yet)

Honesty about the residual debts from Phase 12.3 documented explicitly:

OTel logs bridge: log.X(...) go to stderr (with trace_id propagated). For them to ALSO go to the OTel log signal of the backend (correlated with spans there), you have to wait for sub-step 12.3.iter2.b — designed but not shipped.
OTel metrics bridge: the metrics that @metric emits dispatch to the Prometheus recorder when @server(prometheus=true). For them to also go to the OTel metrics signal (push to Honeycomb metrics, NewRelic metrics) you need to wait for the upstream crate metrics-exporter-opentelemetry release compatible with opentelemetry_sdk 0.32 (documented debt, not laziness on our part).
Tail-based sampling: only head-based (TraceIdRatioBased). For "export traces that had errors" or "sample by latency" you have to run OTel collector in the middle with that config.
Continuous profiling (pyroscope/pprof) — not integrated.
DB auto-instrumentation: the native ORM emits the executed SQL inside the calling handler's span. For queries via raw db.query(...), today no child span is created (minor debt).

Closing

Observability in Python is the area where "library-first" most visibly pays off poorly: each signal lives in a different lib, each lib has its own config, each handler needs boilerplate to use them, and keeping consistency across the three signals is your responsibility.

Fitz puts the three signals in the language, with auto-instrumentation for what every app needs (HTTP spans + access logs + metrics), optional decorators for custom things (@trace/@metric), and native connection between signals (trace_id propagated, Secret redacted).

The target remains the OTel ecosystem — you export to the same backends, with the same standard env vars. What changes is your code.

If your observability pipeline is already stabilized with OTel in FastAPI and works, there's no urgency. If you're starting a project in 2026 and want to avoid the day of plumbing, it's worth a try.

Next post in the series: "Typed ORM with automatic migrations: Fitz vs SQLAlchemy + Alembic + Pydantic" — the last piece of the stack, with reproducible benchmarks.

Repo: github.com/Thegreekman76/fitz
Chapter 33 of the guide (Observability): docs/guide.md

Buscando 5 beta testers para FitzWatch — el status page que construí en Fitz

Martin Palopoli — Mon, 29 Jun 2026 12:19:57 +0000

TL;DR

Construí fitzwatch (https://www.fitzwatch.com), un status page para
apps en prod estilo Statuspage de Atlassian pero más liviano y barato.
El producto está vivo y operacional. Antes de abrirlo al público y
sumar planes pagos, busco 5 beta testers para validar el producto
con users externos.

Si tenés algo en prod que monitorees (o que deberías monitorear) y te
interesa probar algo nuevo + dar feedback directo, leé abajo.

Qué hace fitzwatch

Monitorea endpoints HTTP/TCP cada 10s.
Dispara incidents con lifecycle (investigating → identified → monitoring → resolved).
Status page público scoped por slug /p/<tu-slug>/ con branding propio (logo, color, nombre).
Notificaciones por email y webhook cuando algo se cae.
RSS feed + widget embebible para tu site.
90-day uptime chart, subscribers públicos, multi-tenant.

Comparado con Statuspage
(~$29/mes) o Better Stack (~$24/mes), apunta
al sweet spot del founder de side-project o micro-SaaS al que le
importa el uptime pero no quiere pagar tooling enterprise para 1-3
endpoints.

Por qué construí esto en Fitz (mi propio lenguaje)

Fitz es un lenguaje que vengo
desarrollando hace meses con HTTP/async/ORM como ciudadanos de primera
clase del core del lenguaje. fitzwatch es la prueba de fuego: ¿puedo
construir un producto real, complejo, dockerizable, deployable a prod,
usando solo Fitz?

Spoiler: sí. ~22 módulos .fitz, scheduler nativo con @cron, ORM
declarativo sobre Postgres, OpenAPI auto, WebSockets tipados, paridad
bit-a-bit fitz run ↔ fitz build a binario nativo en Docker.

(Eso es post separado — acá hablamos de la beta.)

La propuesta

Si te sumás:

6 meses de acceso gratis ilimitado.
Soporte directo por email cuando lo necesites.

Te pido a cambio:

Setup de 1 monitor real (~5min).
Call de 20min de feedback al día 7-10.
5 preguntas por email al día 14 (~5min).

Si te copa el producto, opcional: permiso para usarte como testimonial.

Cómo sumarte

Dejame un comentario abajo, mandame DM acá en dev.to. Te respondo en el día.

Stack del producto si te da curiosidad:

Backend: Fitz (compilado a binario Rust nativo via codegen).
DB: Postgres 16 + ORM declarativo de Fitz.
Frontend público: vanilla JS + Tailwind.
Frontend admin: Vue 3 + Vuetify + Chart.js (CDN, sin build step).
Deploy: Docker + nginx en VPS DigitalOcean.

Gracias por leer 🙏

Auth con JWT, RBAC y token blacklist sin pegar 5 librerías: Fitz vs FastAPI + python-jose + passlib + Redis + RBAC casero

Martin Palopoli — Fri, 26 Jun 2026 10:19:35 +0000

El flow completo de auth — login, hashing Argon2id, JWT firma y verificación, RBAC custom apilable, logout con blacklist persistente — lado a lado en FastAPI y en Fitz. Cero librerías nuevas, todo built-in al lenguaje.

La auth de verdad necesita más cosas de las que uno cree

Empezás con "necesito login con JWT". Después aparece:

Hashing de passwords con un algoritmo bueno (Argon2id, no bcrypt para proyectos nuevos).
RBAC porque no alcanza con "logueado/no logueado" — hay roles, hay endpoints de admin.
Logout que de verdad invalide el token (no solo "olvidate de él del lado del cliente").
Refresh tokens para no obligar al usuario a re-loguearse cada hora.
Cleanup de la blacklist para no llenar Redis con tokens vencidos.

En FastAPI, cada una de esas cosas es una librería distinta más código pegamento. En Fitz, son parte del lenguaje.

El stack típico de Python

pip install python-jose[cryptography] passlib[argon2] argon2-cffi redis python-multipart

auth.py:

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import redis.asyncio as redis
import uuid
import time

SECRET_KEY = os.environ["JWT_SECRET"]
ALGORITHM = "HS256"

pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
redis_client = redis.from_url(os.environ["REDIS_URL"])

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta = timedelta(hours=1)):
    to_encode = data.copy()
    jti = str(uuid.uuid4())
    exp = datetime.now(tz=timezone.utc) + expires_delta
    to_encode.update({"jti": jti, "exp": exp})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def is_blacklisted(jti: str) -> bool:
    return await redis_client.exists(f"blacklist:{jti}") > 0

async def blacklist_token(jti: str, exp: int):
    ttl = exp - int(time.time())
    if ttl > 0:
        await redis_client.set(f"blacklist:{jti}", "1", ex=ttl)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        raise HTTPException(status_code=401, detail="invalid token")
    jti = payload.get("jti")
    if jti and await is_blacklisted(jti):
        raise HTTPException(status_code=401, detail="token revoked")
    user = await db.fetch_user(payload["sub"])
    if not user:
        raise HTTPException(status_code=401, detail="user not found")
    return user

def require_role(*roles):
    async def checker(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(status_code=403, detail="forbidden")
        return user
    return checker

endpoints.py:

@app.post("/login")
async def login(creds: Credentials):
    user = await db.fetch_user_by_email(creds.email)
    if not user or not verify_password(creds.password, user.password_hash):
        raise HTTPException(status_code=401, detail="invalid credentials")
    token = create_access_token({"sub": user.email, "role": user.role})
    return {"token": token}

@app.get("/me")
async def me(user: User = Depends(get_current_user)) -> User:
    return user

@app.get("/admin/users")
async def admin_list(user: User = Depends(require_role("admin"))) -> list[User]:
    return await db.fetch_all_users()

@app.post("/articles")
async def create_article(
    body: ArticleIn,
    user: User = Depends(require_role("editor", "publisher")),
) -> Article:
    return await db.create_article(body, user.id)

@app.post("/auth/logout")
async def logout(user_and_jti: tuple[User, str] = Depends(get_user_and_jti)):
    user, jti = user_and_jti
    payload = jwt.decode(user.token, SECRET_KEY, algorithms=[ALGORITHM])
    await blacklist_token(jti, payload["exp"])
    return {"ok": True}

Ahí va. Cinco librerías, ~100 líneas de boilerplate antes de escribir lógica de negocio, una conexión Redis adicional, y la responsabilidad de mantenerlo sincronizado.

Lo mismo en Fitz

type User { id: Int, email: Str, name: Str, role: Str, password_hash: Str }
type Credentials { email: Str, password: Str }
type LoginResponse { token: Str }
type Article { id: Int, title: Str, body: Str, user_id: Int }
type ArticleIn { title: Str, body: Str }

let SECRET = secret("JWT_SECRET")

@auth_provider
async fn check_token(headers: Map<Str, Str>, db: DbConn) -> Result<User> {
    let raw = match headers.get("authorization") {
        Ok(v) => v,
        Err(_) => return Err("falta header Authorization"),
    }
    let parts = raw.split(" ")
    if (parts.len() != 2 or parts[0] != "Bearer") {
        return Err("se esperaba 'Bearer <token>'")
    }
    let claims = jwt.decode(parts[1], SECRET.expose())?
    let jti = match claims.get("jti") {
        Ok(v) => v,
        Err(_) => return Err("token sin jti"),
    }
    if (auth.is_blacklisted(db, jti).await?) {
        return Err("token revocado")
    }
    return User.where(fn(u) => u.email == claims["sub"]).first(db).await
}

@post("/login")
async fn login(db: DbConn, creds: Credentials) -> LoginResponse {
    let user: User = match User.where(fn(u) => u.email == creds.email).first(db).await {
        Ok(u) => u,
        Err(_) => return 401 { "error": "credenciales inválidas" },
    }
    if (not hash.verify(creds.password, user.password_hash)) {
        return 401 { "error": "credenciales inválidas" }
    }
    let exp = DateTime.now().timestamp() + 86400  // 24 h de validez
    // El payload del JWT es `Map<Str, Str>` en `fitz build` MVP — los
    // numéricos se serializan a string. En `fitz run` la heterogeneidad
    // pasa, pero usamos el shape estricto para mantener paridad bit-a-bit.
    let claims = {
        "sub": user.email,
        "role": user.role,
        "jti": Uuid.v4().to_str(),
        "exp": "{exp}",
    }
    return LoginResponse { token: jwt.encode(claims, SECRET.expose()) }
}

@authenticated @get("/me")
fn me(user: User) -> User => user

@admin @get("/admin/users")
async fn admin_list(db: DbConn, user: User) -> List<User> {
    return User.all(db).await
}

@requires("editor") @requires("publisher")
@post("/articles")
async fn create_article(db: DbConn, body: ArticleIn, user: User) -> Article {
    return Article.insert(db, Article { id: 0, title: body.title, body: body.body, user_id: user.id }).await?
}

@authenticated @post("/auth/logout")
async fn logout(db: DbConn, user: User, headers: Map<Str, Str>) -> Map<Str, Bool> {
    // Re-decodificamos el token sólo para sacar jti del claims.
    let raw = headers["authorization"]
    let token = raw.split(" ")[1]
    let claims = jwt.decode(token, SECRET.expose())?
    let jti = claims.get("jti")?
    // TTL fresca de 24h. Cuando el JWT expira, `jwt.decode` ya lo rechaza
    // por expirado; la blacklist solo previene reuse del jti antes de exp.
    // No leemos exp del payload (vendría como Str y `auth.blacklist` exige Int).
    let ttl = DateTime.now().timestamp() + 86400
    auth.blacklist(db, jti, ttl).await?
    return { "ok": true }
}

@cron("0 0 3 * * *")
async fn cleanup_blacklist(db: DbConn) {
    auth.cleanup_expired(db).await
}

Cero pip install. Cero archivo auth.py aparte. El compilador valida cada decorador estáticamente.

La tabla cruda

Pieza	Python (FastAPI + 5 libs)	Fitz
Hash de passwords	`passlib[argon2]` + `CryptContext(schemes=["argon2"])`	`hash.password(...)` built-in
Verificar password	`pwd_context.verify(...)`	`hash.verify(...)`
Firmar JWT	`jose.jwt.encode(..., SECRET, algorithm="HS256")`	`jwt.encode(claims, SECRET)`
Decodificar JWT	`jose.jwt.decode(...)` + try/except	`jwt.decode(token, SECRET)?`
Auth scheme	`OAuth2PasswordBearer` + `Depends(...)`	`@auth_provider fn check_token(...)`
Endpoint protegido	`Depends(get_current_user)` en cada uno	`@authenticated`
Admin only	RBAC casero con `require_role("admin")`	`@admin`
RBAC custom	Helper `require_role(*roles)` + Depends	`@requires("editor")` apilable
Token blacklist	Redis + lógica manual + TTL calculado a mano	`auth.blacklist(db, jti, exp)`
Cleanup vencidos	Redis SET TTL (auto) o tú manual si DB	`auth.cleanup_expired(db)` + `@cron`
Validación estática	Ninguna — error en runtime cuando llega req	`@authenticated` sin `@auth_provider` → error de compilación
Schema OpenAPI con `bearerAuth`	Tenés que declararlo en `app = FastAPI(...)`	Automático cuando hay `@auth_provider`

Por partes

Hash de passwords

Python:

from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")

hashed = pwd_context.hash("password123")
ok = pwd_context.verify("password123", hashed)

Fitz:

let hashed = hash.password("password123")
let ok = hash.verify("password123", hashed)

Argon2id por defecto (recomendación OWASP). Sin elegir esquemas, sin configurar deprecated="auto". Lo que querés con buenos defaults.

JWT

Python:

from jose import jwt, JWTError
from datetime import datetime, timezone, timedelta

exp = datetime.now(tz=timezone.utc) + timedelta(hours=24)
token = jwt.encode({"sub": "ada", "exp": exp}, SECRET, algorithm="HS256")

try:
    claims = jwt.decode(token, SECRET, algorithms=["HS256"])
except JWTError as e:
    raise HTTPException(401, f"invalid token: {e}")

Fitz:

let exp = DateTime.now().timestamp() + 86400
let token = jwt.encode({ "sub": "ada", "exp": "{exp}" }, SECRET)

match jwt.decode(token, SECRET) {
    Ok(claims) => process(claims),
    Err(e) => return 401 { "error": "{e}" },
}

HS256 por default. HS384/HS512 disponibles con kwarg. Excepción → Result::Err automático, manejado con ? o match.

RBAC custom apilable

Acá es donde Fitz claramente diferencia. En FastAPI, RBAC más allá de "admin sí / no" significa escribir un helper:

def require_role(*roles):
    async def checker(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(403, "forbidden")
        return user
    return checker

@app.post("/articles")
async def create_article(
    body: ArticleIn,
    user: User = Depends(require_role("editor", "publisher")),
):
    ...

En Fitz:

@requires("editor") @requires("publisher")
@post("/articles")
async fn create_article(db: DbConn, body: ArticleIn, user: User) -> Article {
    ...
}

Multi-decorator es OR (el user pertenece a uno u otro role). El checker valida estáticamente:

Que User tenga campo role: Str no nullable.
Que no haya duplicados (@requires("editor") @requires("editor") → error compile-time).
Que el handler tenga un @auth_provider activo en el archivo.
Que el último param del handler que matchee con el type del provider sea el inyectado.

Errores que en Python aparecen en runtime cuando un user le pega al endpoint, en Fitz aparecen en fitz check antes de commitear.

Token blacklist persistente

Esta es la pieza que muchos proyectos en FastAPI nunca terminan de hacer bien — porque significa sumar Redis aunque tu app no la necesite para nada más, escribir el client async, calcular TTLs a mano, decidir si la falla de Redis tira el request o lo deja pasar...

async def blacklist_token(jti: str, exp: int):
    ttl = exp - int(time.time())
    if ttl > 0:
        await redis_client.set(f"blacklist:{jti}", "1", ex=ttl)

async def is_blacklisted(jti: str) -> bool:
    return await redis_client.exists(f"blacklist:{jti}") > 0

Plus mantener Redis vivo, montar el client, manejar reconexiones.

En Fitz, usás el Postgres que ya tenés:

@authenticated
@post("/auth/logout")
async fn logout(db: DbConn, user: User) -> Map<Str, Bool> {
    auth.blacklist(db, user.jti, user.exp).await?
    return { "ok": true }
}

@cron("0 0 3 * * *")  // 3 AM diario
async fn cleanup_blacklist(db: DbConn) {
    auth.cleanup_expired(db).await
}

Tabla fitz_token_blacklist(jti TEXT PRIMARY KEY, expires_at BIGINT NOT NULL) se auto-crea con CREATE TABLE IF NOT EXISTS al primer call. Auto-filtro expires_at > now() cuando checkea (tokens vencidos no necesitan seguir bloqueando — jwt.decode los rechaza primero por exp).

El patrón /auth/logout y /auth/refresh no se auto-monta (intencional — el flow exacto varía por proyecto), pero los tres builtins (auth.blacklist/auth.is_blacklisted/auth.cleanup_expired) te lo dan hecho en ~10 líneas de tu código.

OpenAPI con `bearerAuth` automático

En FastAPI tenés que declararlo:

app = FastAPI(
    swagger_ui_init_oauth={...},
)
# y la security scheme en cada endpoint manualmente o con dependencies

En Fitz, cuando hay @auth_provider en el programa, el schema OpenAPI se emite con:

{
  "components": {
    "securitySchemes": {
      "bearerAuth": {
        "type": "http",
        "scheme": "bearer",
        "bearerFormat": "JWT"
      }
    }
  },
  "paths": {
    "/me": {
      "get": {
        "security": [{"bearerAuth": []}],
        "responses": {
          "200": {...},
          "401": {"description": "auth"},
        }
      }
    }
  }
}

La UI de Scalar en /docs te muestra el botón "Authorize" funcional. Auto-mounted. Sin configurar.

El checker hace gran parte del trabajo

Antes de levantar el server, fitz check valida:

@authenticated sin @auth_provider → "se usa @authenticated pero no hay @auth_provider declarado".
@admin sobre handler cuyo User no tiene role: Str no nullable → "para usar @admin el type User debe tener campo role: Str".
@requires("X") con typo → no chequea valores (el role es un Str arbitrario), pero detecta apilados duplicados.
JWT decode return type → Result<Map<Str, Str>>, exige manejo de Err.
El param inyectado por @auth_provider → su tipo debe matchear el del User retornado por el provider.

Errores que en FastAPI/Flask típicamente se descubren cuando el request llega a producción, en Fitz aparecen en CI.

Decisiones de diseño que vale la pena entender

`auth.blacklist` exige DbConn explícita

En lugar de un singleton global, los 3 builtins (auth.blacklist/is_blacklisted/cleanup_expired) toman DbConn como primer arg. Razón: hace explícito el costo en cada call site, y testeable contra una DB de test.

Mensaje de 403 enriquecido para `@requires`

Cuando el role no matchea:

{
  "error": "forbidden",
  "current_role": "viewer",
  "required_roles": ["editor", "publisher"]
}

Habilita debugging serio y observability (los logs estructurados con log.info("auth.denied", { user_id, role, required }) te dicen exactamente qué falló).

`@requires` implica auth

No necesitás apilar @authenticated @requires("editor") — el @requires ya corre el provider. Apilarlo es no-op (y el checker lo flagea).

Lo que Fitz NO te da (todavía)

OAuth2 social login (Google/GitHub/etc.) — el flow OAuth contra el provider lo hacés a mano contra los endpoints conocidos.
Multi-role (user.roles: List<Str>) — hoy es single role. Roles compuestos: deuda residual de 9.w.1.iter2.
Asymmetric JWT (RS256/ES256 con PEM y rotación) — solo HS256/384/512.
Role hierarchy (admin > editor > viewer) — no se modela. Workaround: si admin debe poder hacer lo que editor, @requires("admin") @requires("editor") (apilado).
Session cookie-based alternativo a JWT.
/auth/refresh y /auth/logout auto-mounted — el patrón canónico (~10 LoC con los builtins) está en el cap 28 de la guía pero queda manual. Auto-mount con @server(auto_auth_endpoints=true) es deuda visible.

Cierre

Lo que más me cansa de hacer auth en Python no es escribir el primer endpoint — es mantener consistencia entre tres lugares: el Depends(...) de FastAPI, la lib de JWT, y el RBAC casero. Cuando agregás @requires("editor") y te olvidaste de actualizar el helper del RBAC, no te enterás hasta que un user te reporta un bug.

En Fitz el checker estático cierra ese loop. El @auth_provider es la fuente de verdad del tipo User. @authenticated/@admin/@requires consumen ese tipo. La OpenAPI lo refleja. La blacklist usa la misma DB. Cero deps externas para auth.

Si tu proyecto está en uno de los casos del MVP — JWT, roles simples, blacklist en Postgres — Fitz te ahorra el día de pegar librerías. Si necesitás OAuth social y multi-role, todavía es Python (o tomalo como invitación a abrir un issue).

Próximo post de la serie: "Tracing distribuido, métricas Prometheus y logs estructurados con dos decoradores: @trace/@metric en Fitz vs el setup de OpenTelemetry en FastAPI" — observability lado a lado.

Repo: github.com/Thegreekman76/fitz
Capítulo 28 de la guía (Auth nativa): docs/guide.md

JWT auth, RBAC, and token blacklist without gluing 5 libraries: Fitz vs FastAPI + python-jose + passlib + Redis + home-grown RBAC

Martin Palopoli — Fri, 26 Jun 2026 10:19:17 +0000

The full auth flow — login, Argon2id hashing, JWT sign and verify, stackable custom RBAC, logout with persistent blacklist — side by side in FastAPI and in Fitz. Zero new libraries, everything built into the language.

Real auth needs more pieces than you'd think

You start with "I need login with JWT". Then comes:

Password hashing with a good algorithm (Argon2id, not bcrypt for new projects).
RBAC because "logged in / not logged in" isn't enough — there are roles, there are admin endpoints.
Logout that actually invalidates the token (not just "forget about it on the client side").
Refresh tokens so users don't have to re-log every hour.
Cleanup of the blacklist so you don't fill Redis with expired tokens.

In FastAPI, each of those is a separate library plus glue code. In Fitz, they're part of the language.

The typical Python stack

pip install python-jose[cryptography] passlib[argon2] argon2-cffi redis python-multipart

auth.py:

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import redis.asyncio as redis
import uuid
import time

SECRET_KEY = os.environ["JWT_SECRET"]
ALGORITHM = "HS256"

pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
redis_client = redis.from_url(os.environ["REDIS_URL"])

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta = timedelta(hours=1)):
    to_encode = data.copy()
    jti = str(uuid.uuid4())
    exp = datetime.now(tz=timezone.utc) + expires_delta
    to_encode.update({"jti": jti, "exp": exp})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def is_blacklisted(jti: str) -> bool:
    return await redis_client.exists(f"blacklist:{jti}") > 0

async def blacklist_token(jti: str, exp: int):
    ttl = exp - int(time.time())
    if ttl > 0:
        await redis_client.set(f"blacklist:{jti}", "1", ex=ttl)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        raise HTTPException(status_code=401, detail="invalid token")
    jti = payload.get("jti")
    if jti and await is_blacklisted(jti):
        raise HTTPException(status_code=401, detail="token revoked")
    user = await db.fetch_user(payload["sub"])
    if not user:
        raise HTTPException(status_code=401, detail="user not found")
    return user

def require_role(*roles):
    async def checker(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(status_code=403, detail="forbidden")
        return user
    return checker

endpoints.py:

@app.post("/login")
async def login(creds: Credentials):
    user = await db.fetch_user_by_email(creds.email)
    if not user or not verify_password(creds.password, user.password_hash):
        raise HTTPException(status_code=401, detail="invalid credentials")
    token = create_access_token({"sub": user.email, "role": user.role})
    return {"token": token}

@app.get("/me")
async def me(user: User = Depends(get_current_user)) -> User:
    return user

@app.get("/admin/users")
async def admin_list(user: User = Depends(require_role("admin"))) -> list[User]:
    return await db.fetch_all_users()

@app.post("/articles")
async def create_article(
    body: ArticleIn,
    user: User = Depends(require_role("editor", "publisher")),
) -> Article:
    return await db.create_article(body, user.id)

@app.post("/auth/logout")
async def logout(user_and_jti: tuple[User, str] = Depends(get_user_and_jti)):
    user, jti = user_and_jti
    payload = jwt.decode(user.token, SECRET_KEY, algorithms=[ALGORITHM])
    await blacklist_token(jti, payload["exp"])
    return {"ok": True}

There it is. Five libraries, ~100 lines of boilerplate before you write any business logic, an extra Redis connection, and the responsibility to keep it in sync.

The same thing in Fitz

type User { id: Int, email: Str, name: Str, role: Str, password_hash: Str }
type Credentials { email: Str, password: Str }
type LoginResponse { token: Str }
type Article { id: Int, title: Str, body: Str, user_id: Int }
type ArticleIn { title: Str, body: Str }

let SECRET = secret("JWT_SECRET")

@auth_provider
async fn check_token(headers: Map<Str, Str>, db: DbConn) -> Result<User> {
    let raw = match headers.get("authorization") {
        Ok(v) => v,
        Err(_) => return Err("missing Authorization header"),
    }
    let parts = raw.split(" ")
    if (parts.len() != 2 or parts[0] != "Bearer") {
        return Err("expected 'Bearer <token>'")
    }
    let claims = jwt.decode(parts[1], SECRET.expose())?
    let jti = match claims.get("jti") {
        Ok(v) => v,
        Err(_) => return Err("token missing jti"),
    }
    if (auth.is_blacklisted(db, jti).await?) {
        return Err("token revoked")
    }
    return User.where(fn(u) => u.email == claims["sub"]).first(db).await
}

@post("/login")
async fn login(db: DbConn, creds: Credentials) -> LoginResponse {
    let user: User = match User.where(fn(u) => u.email == creds.email).first(db).await {
        Ok(u) => u,
        Err(_) => return 401 { "error": "invalid credentials" },
    }
    if (not hash.verify(creds.password, user.password_hash)) {
        return 401 { "error": "invalid credentials" }
    }
    let exp = DateTime.now().timestamp() + 86400  // 24h validity
    // The JWT payload is `Map<Str, Str>` in the `fitz build` MVP — numerics
    // serialize to string. In `fitz run` heterogeneity passes, but we use
    // the strict shape to preserve bit-for-bit parity.
    let claims = {
        "sub": user.email,
        "role": user.role,
        "jti": Uuid.v4().to_str(),
        "exp": "{exp}",
    }
    return LoginResponse { token: jwt.encode(claims, SECRET.expose()) }
}

@authenticated @get("/me")
fn me(user: User) -> User => user

@admin @get("/admin/users")
async fn admin_list(db: DbConn, user: User) -> List<User> {
    return User.all(db).await
}

@requires("editor") @requires("publisher")
@post("/articles")
async fn create_article(db: DbConn, body: ArticleIn, user: User) -> Article {
    return Article.insert(db, Article { id: 0, title: body.title, body: body.body, user_id: user.id }).await?
}

@authenticated @post("/auth/logout")
async fn logout(db: DbConn, user: User, headers: Map<Str, Str>) -> Map<Str, Bool> {
    // We re-decode the token just to pull jti from the claims.
    let raw = headers["authorization"]
    let token = raw.split(" ")[1]
    let claims = jwt.decode(token, SECRET.expose())?
    let jti = claims.get("jti")?
    // Fresh 24h TTL. When the JWT expires, `jwt.decode` rejects it as
    // expired; the blacklist only prevents reuse of the jti before exp.
    // We don't read exp from the payload (it would come back as Str and
    // `auth.blacklist` requires Int).
    let ttl = DateTime.now().timestamp() + 86400
    auth.blacklist(db, jti, ttl).await?
    return { "ok": true }
}

@cron("0 0 3 * * *")
async fn cleanup_blacklist(db: DbConn) {
    auth.cleanup_expired(db).await
}

Zero pip install. Zero separate auth.py file. The compiler validates every decorator statically.

The raw table

Piece	Python (FastAPI + 5 libs)	Fitz
Password hashing	`passlib[argon2]` + `CryptContext(schemes=["argon2"])`	`hash.password(...)` built-in
Verify password	`pwd_context.verify(...)`	`hash.verify(...)`
Sign JWT	`jose.jwt.encode(..., SECRET, algorithm="HS256")`	`jwt.encode(claims, SECRET)`
Decode JWT	`jose.jwt.decode(...)` + try/except	`jwt.decode(token, SECRET)?`
Auth scheme	`OAuth2PasswordBearer` + `Depends(...)`	`@auth_provider fn check_token(...)`
Protected endpoint	`Depends(get_current_user)` on each one	`@authenticated`
Admin only	Home-grown RBAC with `require_role("admin")`	`@admin`
Custom RBAC	`require_role(*roles)` helper + Depends	`@requires("editor")` stackable
Token blacklist	Redis + manual logic + manually computed TTL	`auth.blacklist(db, jti, exp)`
Expired cleanup	Redis SET TTL (auto) or manual if DB	`auth.cleanup_expired(db)` + `@cron`
Static validation	None — runtime error when a request hits	`@authenticated` without `@auth_provider` → compile error
OpenAPI `bearerAuth` schema	Declare in `app = FastAPI(...)`	Automatic when `@auth_provider` is present

Piece by piece

Password hashing

Python:

from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")

hashed = pwd_context.hash("password123")
ok = pwd_context.verify("password123", hashed)

Fitz:

let hashed = hash.password("password123")
let ok = hash.verify("password123", hashed)

Argon2id by default (OWASP recommendation). No picking schemes, no configuring deprecated="auto". What you want with good defaults.

JWT

Python:

from jose import jwt, JWTError
from datetime import datetime, timezone, timedelta

exp = datetime.now(tz=timezone.utc) + timedelta(hours=24)
token = jwt.encode({"sub": "ada", "exp": exp}, SECRET, algorithm="HS256")

try:
    claims = jwt.decode(token, SECRET, algorithms=["HS256"])
except JWTError as e:
    raise HTTPException(401, f"invalid token: {e}")

Fitz:

let exp = DateTime.now().timestamp() + 86400
let token = jwt.encode({ "sub": "ada", "exp": "{exp}" }, SECRET)

match jwt.decode(token, SECRET) {
    Ok(claims) => process(claims),
    Err(e) => return 401 { "error": "{e}" },
}

HS256 by default. HS384/HS512 available via kwarg. Exception → Result::Err automatic, handled with ? or match.

Stackable custom RBAC

This is where Fitz clearly stands apart. In FastAPI, RBAC beyond "admin yes / no" means writing a helper:

def require_role(*roles):
    async def checker(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(403, "forbidden")
        return user
    return checker

@app.post("/articles")
async def create_article(
    body: ArticleIn,
    user: User = Depends(require_role("editor", "publisher")),
):
    ...

In Fitz:

@requires("editor") @requires("publisher")
@post("/articles")
async fn create_article(db: DbConn, body: ArticleIn, user: User) -> Article {
    ...
}

Multi-decorator is OR (the user belongs to one or the other role). The checker validates statically:

That User has a non-nullable role: Str field.
That there are no duplicates (@requires("editor") @requires("editor") → compile-time error).
That the handler has an active @auth_provider in the file.
That the handler's last param matching the provider's return type is the injected one.

Errors that in Python appear at runtime when a user hits the endpoint, in Fitz appear in fitz check before you commit.

Persistent token blacklist

This is the piece many FastAPI projects never finish properly — because it means adding Redis even though your app doesn't need it for anything else, writing the async client, computing TTLs by hand, deciding whether a Redis failure should fail the request or let it through...

async def blacklist_token(jti: str, exp: int):
    ttl = exp - int(time.time())
    if ttl > 0:
        await redis_client.set(f"blacklist:{jti}", "1", ex=ttl)

async def is_blacklisted(jti: str) -> bool:
    return await redis_client.exists(f"blacklist:{jti}") > 0

Plus keeping Redis alive, mounting the client, handling reconnects.

In Fitz, you use the Postgres you already have:

@authenticated
@post("/auth/logout")
async fn logout(db: DbConn, user: User) -> Map<Str, Bool> {
    auth.blacklist(db, user.jti, user.exp).await?
    return { "ok": true }
}

@cron("0 0 3 * * *")  // 3 AM daily
async fn cleanup_blacklist(db: DbConn) {
    auth.cleanup_expired(db).await
}

The fitz_token_blacklist(jti TEXT PRIMARY KEY, expires_at BIGINT NOT NULL) table auto-creates with CREATE TABLE IF NOT EXISTS on the first call. Auto-filters expires_at > now() on check (expired tokens don't need to keep blocking — jwt.decode rejects them first via exp).

The /auth/logout and /auth/refresh patterns are not auto-mounted (intentional — the exact flow varies by project), but the three builtins (auth.blacklist/auth.is_blacklisted/auth.cleanup_expired) give you the work in ~10 lines of your code.

Automatic `bearerAuth` in OpenAPI

In FastAPI you have to declare it:

app = FastAPI(
    swagger_ui_init_oauth={...},
)
# and the security scheme on each endpoint manually or via dependencies

In Fitz, when the program has @auth_provider, the OpenAPI schema emits:

{
  "components": {
    "securitySchemes": {
      "bearerAuth": {
        "type": "http",
        "scheme": "bearer",
        "bearerFormat": "JWT"
      }
    }
  },
  "paths": {
    "/me": {
      "get": {
        "security": [{"bearerAuth": []}],
        "responses": {
          "200": {...},
          "401": {"description": "auth"}
        }
      }
    }
  }
}

The Scalar UI at /docs shows you a working "Authorize" button. Auto-mounted. No config.

The checker does much of the work

Before bringing up the server, fitz check validates:

@authenticated without @auth_provider → "uses @authenticated but no @auth_provider declared".
@admin over a handler whose User has no role: Str non-nullable → "to use @admin the User type must have a role: Str field".
@requires("X") with a typo → doesn't check values (role is an arbitrary Str), but detects duplicate stacks.
JWT decode return type → Result<Map<Str, Str>>, requires handling Err.
The param injected by @auth_provider → its type must match the User returned by the provider.

Errors that in FastAPI/Flask typically get discovered when the request lands in production, in Fitz show up in CI.

Design decisions worth understanding

`auth.blacklist` requires an explicit DbConn

Instead of a global singleton, the 3 builtins (auth.blacklist/is_blacklisted/cleanup_expired) take DbConn as the first arg. Reason: it makes the cost explicit at each call site, and testable against a test DB.

Enriched 403 message for `@requires`

When the role doesn't match:

{
  "error": "forbidden",
  "current_role": "viewer",
  "required_roles": ["editor", "publisher"]
}

Enables serious debugging and observability (structured logs with log.info("auth.denied", { user_id, role, required }) tell you exactly what failed).

`@requires` implies auth

You don't need to stack @authenticated @requires("editor") — @requires already runs the provider. Stacking it is a no-op (and the checker flags it).

What Fitz does NOT give you (yet)

OAuth2 social login (Google/GitHub/etc.) — the OAuth flow against the provider you do by hand against the known endpoints.
Multi-role (user.roles: List<Str>) — today it's single role. Composite roles: residual debt of 9.w.1.iter2.
Asymmetric JWT (RS256/ES256 with PEM and rotation) — only HS256/384/512.
Role hierarchy (admin > editor > viewer) — not modeled. Workaround: if admin should also be able to do what editor does, @requires("admin") @requires("editor") (stacked).
Cookie-based sessions as an alternative to JWT.
/auth/refresh and /auth/logout auto-mounted — the canonical pattern (~10 LoC with the builtins) is in chapter 28 of the guide but stays manual. Auto-mount with @server(auto_auth_endpoints=true) is visible debt.

Closing

What tires me most about doing auth in Python isn't writing the first endpoint — it's keeping consistency across three places: FastAPI's Depends(...), the JWT library, and the home-grown RBAC. When you add @requires("editor") and you forgot to update the RBAC helper, you don't find out until a user reports a bug.

In Fitz the static checker closes that loop. The @auth_provider is the source of truth for the User type. @authenticated/@admin/@requires consume that type. OpenAPI reflects it. The blacklist uses the same DB. Zero external deps for auth.

If your project is in one of the MVP cases — JWT, simple roles, blacklist in Postgres — Fitz saves you the day of gluing libraries together. If you need OAuth social and multi-role, it's still Python (or take it as an invitation to open an issue).

Next post in the series: "Distributed tracing, Prometheus metrics, and structured logs with two decorators: @trace/@metric in Fitz vs the OpenTelemetry setup in FastAPI" — observability side by side.

Repo: github.com/Thegreekman76/fitz
Chapter 28 of the guide (Native auth): docs/guide.md

Cron jobs sin Celery, sin Redis, sin Beat: cómo Fitz mete un scheduler distribuido adentro del lenguaje

Martin Palopoli — Mon, 22 Jun 2026 23:10:31 +0000

Para correr una task cada 5 minutos en Python necesitás Celery + Redis + Celery Beat + worker process + Dockerfile dedicado. En Fitz es un decorador. Con retry, timezone, persistencia y catch-up adentro del lenguaje.

La historia de siempre

Tu cliente está contento con la API. Está corriendo. Entonces te pide tres cosas:

"¿Podemos limpiar sesiones vencidas cada noche?"
"¿Podés mandar el email de bienvenida en background así el signup no espera?"
"El reporte diario tiene que correr a las 9 AM hora de Buenos Aires."

Tres pedidos chicos. Si estás en Python, la respuesta honesta es: "ok, dame dos días para meter Celery."

El stack típico de Python

pip install celery redis

celery_app.py:

from celery import Celery
from celery.schedules import crontab

celery_app = Celery(
    "myapp",
    broker="redis://redis:6379/0",
    backend="redis://redis:6379/1",
)

celery_app.conf.beat_schedule = {
    "cleanup-sessions-nightly": {
        "task": "myapp.tasks.cleanup_old_sessions",
        "schedule": crontab(hour=3, minute=0),
    },
    "daily-report": {
        "task": "myapp.tasks.generate_daily_report",
        "schedule": crontab(hour=12, minute=0),  # 9am Buenos Aires = 12:00 UTC
        # ojo: si DST cambia, esto te llega tarde
    },
}
celery_app.conf.timezone = "UTC"

tasks.py:

from .celery_app import celery_app
from .db import get_session

@celery_app.task(autoretry_for=(Exception,), retry_backoff=True, max_retries=3)
def cleanup_old_sessions():
    with get_session() as s:
        s.execute("DELETE FROM sessions WHERE expires_at < now()")
        s.commit()

@celery_app.task(autoretry_for=(SMTPError,), retry_backoff=True, max_retries=5)
def send_welcome_email(email: str):
    smtp.send(email, "Welcome!")

api.py:

@app.post("/signup")
def signup(creds: Credentials):
    user = create_user(creds)
    send_welcome_email.delay(user.email)  # fire-and-forget vía broker
    return user

docker-compose.yml:

services:
  api:
    build: .
    command: uvicorn api:app --host 0.0.0.0
  celery-worker:
    build: .
    command: celery -A celery_app worker --loglevel=info
    depends_on: [redis]
  celery-beat:
    build: .
    command: celery -A celery_app beat --loglevel=info
    depends_on: [redis]
  redis:
    image: redis:7-alpine

Mas:

Un supervisord.conf o systemd unit que mantenga viva la cosa cuando crashee.
(Opcional) flower para visibility — 4to proceso.
Conversiones de timezone hechas a mano (Celery beat trabaja en UTC, vos tenés que calcular el offset).
Cuando el worker crashea entre que pegó al broker y completó la task: ¿se reintenta? ¿queda colgada? Depende de cómo te configuraste el acks_late y task_reject_on_worker_lost.

Cuatro procesos en producción. Tres librerías nuevas. Un broker. Convenciones nuevas. Un día de setup.

Lo mismo en Fitz

@cron("0 3 * * *", tz="UTC")
async fn cleanup_old_sessions(db: DbConn) {
    db.exec("DELETE FROM sessions WHERE expires_at < now()").await
}

@cron("0 9 * * *", tz="America/Argentina/Buenos_Aires")
async fn daily_report(db: DbConn) {
    // ...
}

@background
async fn send_welcome_email(email: Str) {
    // cosa cara
}

@post("/signup")
fn signup(creds: Credentials) -> User {
    let user = create_user(creds)
    spawn(send_welcome_email(user.email))  // fire-and-forget tipado
    return user
}

docker-compose.yml:

services:
  api:
    build: .

Eso es todo. Un binario. Un proceso. Sin broker. Sin worker dedicado. Sin beat. El scheduler corre adentro del proceso de Fitz.

La tabla cruda

Item	Python (Celery + Redis + Beat)	Fitz
Setup inicial	4 archivos + 3 servicios + 1 broker	1 decorador
Schedule	`crontab(hour=3, minute=0)` en config	`@cron("0 3 * * *")`
Timezone	UTC + offset manual	`tz="America/Argentina/Buenos_Aires"`
Retry/backoff	`autoretry_for=(...)` + `retry_backoff=True`	`retry={max=3, backoff="exponential", initial_secs=1, max_secs=30}`
Persistencia de runs	Redis result backend + visualización con Flower	`store=db`, tabla auto-creada en Postgres
Catch-up de runs perdidos	No nativo	`catch_up=true`
Background fire-and-forget	`.delay(arg)` con broker	`spawn(fn_call)` directo
Type checking de args	Ninguno (todo serializado vía JSON)	Static —`spawn` exige `@background` y refina a `Future<T>`
Procesos en prod	api + worker + beat + redis	api
Imagen Docker	3 (api, worker, beat) + redis	1

Persistencia opt-in con `store=db`

Cuando querés history de runs para auditoría:

@cron("0 3 * * *", store=db, retry={max=3, backoff="exponential"})
async fn cleanup_old_sessions(db: DbConn) {
    db.exec("DELETE FROM sessions WHERE expires_at < now()").await
}

Al boot, Fitz crea (si no existe) fitz_cron_jobs y fitz_cron_runs. Cada attempt queda persistida con started_at/finished_at/status/attempt/error. Lo consultás con SQL plano:

SELECT job_name, status, attempt, error, started_at
FROM fitz_cron_runs
WHERE started_at > now() - interval '24 hours'
ORDER BY started_at DESC;

Sin instalar Flower. Sin webhook para Sentry. Tu DB que ya tenés.

Catch-up

Si el binario estuvo caído entre las 3 AM y las 7 AM y el cron era a las 3:

Celery beat: la oportunidad se pierde. La task no se vuelve a disparar hasta la próxima medianoche siguiente.
Fitz con catch_up=true: al boot, calcula que hubo un run perdido entre last_run_at y now, ejecuta UN run inmediato (no N — evita spam), y vuelve al schedule normal.

@cron("0 3 * * *", store=db, catch_up=true)
async fn cleanup_old_sessions(db: DbConn) { ... }

Retry con backoff configurable

Tres modos de backoff: exponential (1s, 2s, 4s, 8s...), linear (1s, 2s, 3s, 4s...), constant (1s, 1s, 1s...). Con cap max_secs:

@cron("*/10 * * * *",
    retry={ max=5, backoff="exponential", initial_secs=1, max_secs=60 })
async fn sync_external_api(db: DbConn) {
    // 1s → 2s → 4s → 8s → 16s (capeado a 60s si llegara más alto)
}

En Python con Celery:

@celery_app.task(
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=60,
    retry_jitter=False,
    max_retries=5,
)
def sync_external_api():
    ...

Equivalente. Pero notá que Celery tenés que enumerar las excepciones que disparan retry, y el backoff es un bool en lugar de un kind enum. Fitz hace retry ante cualquier error que devuelva la fn (consistente con el modelo Result del lenguaje), y el modo de backoff es explícito.

Timezone real, no offset hardcoded

Las DST changes son el bug que te despiertan a las 4 AM. Celery beat trabaja en UTC y vos tenés que recordar que entre marzo y noviembre tu cron de las 9 AM Buenos Aires se corre a las 12 UTC, pero el resto del año cambia. Tu cliente está en otro huso. Tu app vende a tres países más.

Fitz acepta IANA timezones directamente:

@cron("0 9 * * *", tz="America/Argentina/Buenos_Aires") async fn buenos_aires() { ... }
@cron("0 9 * * *", tz="America/New_York")              async fn new_york()      { ... }
@cron("0 9 * * *", tz="Asia/Tokyo")                    async fn tokyo()         { ... }

Cada uno corre a las 9 AM local de su tz, con DST manejado por la lib subyacente (chrono-tz). No conversiones a mano.

Background jobs sin broker

# Python
@app.post("/signup")
def signup(creds: Credentials):
    user = create_user(creds)
    send_welcome_email.delay(user.email)  # serializa args → Redis → worker
    return user

// Fitz
@background
async fn send_welcome_email(email: Str) {
    smtp.send(email, "Welcome!")
}

@post("/signup")
fn signup(creds: Credentials) -> User {
    let user = create_user(creds)
    spawn(send_welcome_email(user.email))  // tokio::spawn nativo, tipado
    return user
}

El compilador exige que la fn esté decorada con @background para autorizar el spawn(...) — sin esto, el callsite tira error de tipo en build time. Y refina el retorno a Future<Null> con el tipo concreto del target, no Any. Si send_welcome_email retorna Result<()>, el spawn(...) te lo da tipado.

¿Cuándo NO usar @background y volver a Celery?

Si necesitás que los jobs sobrevivan crashes del proceso → persistencia explícita con store=db cubre cron jobs; para @background con persistencia es deuda residual.
Si necesitás distribuir jobs en N workers en N nodos → Celery con un broker compartido sigue siendo la respuesta. @background corre en el proceso.

Para el 90% de servicios (cleanup nocturno, email transaccional, recálculo de KPIs, sync de cache) el modelo de Fitz alcanza y sobra.

Cron-only mode para systemd

Para servicios que SOLO tienen jobs (sin HTTP), fitz build produce un binario que arranca el scheduler y bloquea con ctrl+c:

@cron("0 3 * * *") async fn cleanup() { ... }
@cron("*/15 * * * *") async fn sync() { ... }

Como systemd unit:

[Unit]
Description=Scheduled jobs for myapp
After=network.target

[Service]
ExecStart=/usr/local/bin/myapp-jobs
Restart=always
User=myapp

[Install]
WantedBy=multi-user.target

Cero brokers. El binario es 5 MB. systemctl restart myapp-jobs y listo.

Paridad bit-a-bit `fitz run` ↔ `fitz build`

Esto es lo que vuelve esto entregable: el binario que producís con fitz build ejecuta el mismo scheduler que fitz run, con el mismo cron expression parser, los mismos retries, la misma timezone. Misma sintaxis, mismas semánticas, sin "ah esto solo funciona en producción si configurás Celery".

Lo que Fitz NO te da (todavía)

Soy honesto sobre dónde no llega:

Distribuir jobs en N workers/nodos compartiendo cola. @cron corre en el proceso. Si corrés dos binarios con el mismo @cron, ambos disparan — bug, no feature. Para horizontal scaling de jobs, sigue siendo Celery (o NATS JetStream, o Temporal). Hay deuda explícita en el roadmap sobre locks distribuidos.
UI tipo Flower. Los datos están en fitz_cron_jobs y fitz_cron_runs. Si querés dashboards, dashboards externos (Grafana, Metabase) cubren.
@background con persistencia entre restarts. Para @cron está en v0.11.2 (cierre 9.w.3.iter2). Para @background arranca el spawn pero no sobrevive crash — diferido a iter3 si entra demanda.
Cancelación de jobs en flight. Hoy no hay API para "cancelar todos los runs en cola de X job". El proceso muere → los runs en flight mueren con él.

Si estás en uno de esos casos, Fitz no es la herramienta hoy. Si no, el modelo de un solo binario cubre el caso real con menos partes móviles.

Cierre

El argumento para sumar Celery a una API en Python típicamente es: "pero después escala mejor." En el 90% de los servicios que escribí en la última década, ese "después" nunca llegó — la app pasó toda su vida útil con menos de 100 jobs por hora y nunca necesitó un cluster de workers.

Para ese 90%, Fitz reemplaza 4 procesos + 3 librerías + 1 broker con un decorador. Cuando llegues al otro 10% donde necesitás scaling horizontal real, Celery sigue ahí — from python import celery también está disponible, podés hacerlo tú mismo.

Pero arrancar el proyecto con el modelo más simple posible y subir la complejidad solo cuando lo necesitás es el ciclo de feedback que Fitz quiere darte.

Próximo post de la serie: "Auth con JWT, RBAC y token blacklist sin pegar 5 librerías: Fitz vs FastAPI + python-jose + passlib + Redis blacklist" — el flow completo de auth, lado a lado.

Repo: github.com/Thegreekman76/fitz
Capítulo 30 de la guía (Jobs sin Celery): docs/guide.md

DEV Community: Martin Palopoli

SMTP outbound as a first-class citizen: send emails without pip install yagmail

The detail that gets forgotten

The typical Python stack

The typical JS/Node stack

The typical Rust stack

The typical Java stack

The Fitz stack

What smtp.send guarantees you

1. Zero external deps

2. Bit-for-bit parity fitz run ↔ fitz build

3. Async from day one

4. Result<T> as error model

5. Magic-link auth in 30 lines

Local setup with MailHog

Honest trade-offs (MVP)

Closing

SMTP outbound como ciudadano de primera clase: mandá emails sin pip install yagmail

El detalle que se olvida

El stack típico de Python

El stack típico de JS/Node

El stack típico de Rust

El stack típico de Java

El stack de Fitz

Lo que smtp.send te garantiza

1. Cero deps externas

2. Paridad bit-a-bit fitz run ↔ fitz build

3. Async desde día uno

4. Result<T> como modelo de error

5. Magic-link auth en 30 líneas

Setup local con MailHog

Trade-offs honestos (MVP)

Cierre

HTTP client as a first-class citizen: zero deps, bit-for-bit parity, async by default

The forgotten detail

The typical Python stack

The typical JS/TS stack

The typical Rust stack

Same thing in Fitz

The raw table

The 6 builtins, all on one page

Error handling: the important delta

Canonical webhook dispatcher (the real case)

fitzwatch-style health checker

Why it matters

What's next

Try it

HTTP client outbound como ciudadano de primera clase: cero deps, paridad bit-a-bit, async natural

El detalle que se olvida

El stack típico de Python

El stack típico de JS/TS

El stack típico de Rust

Lo mismo en Fitz

La tabla cruda

Los 6 builtins, todos en una página

Manejo de errores: el delta importante

Webhook dispatcher canónico (el caso real)

Health checker estilo fitzwatch

Por qué importa

Lo que viene

Probarlo

ORM tipado con migraciones automáticas: Fitz vs SQLAlchemy + Alembic + Pydantic

El doble (triple) tipado del stack Python

Lo mismo en Fitz

La tabla cruda

Por partes

Closure-to-SQL en compile-time

Operadores Postgres nativos

Relations + eager loading

Transactions

Migraciones declarativas

¿Cuán rápido es? — bench reproducible

Lo que Fitz NO te da (todavía)

Escape hatch: SQL crudo cuando lo necesites

Cierre

Typed ORM with automatic migrations: Fitz vs SQLAlchemy + Alembic + Pydantic

The double (triple) typing of the Python stack

The same thing in Fitz

The raw table

Piece by piece

What `smtp.send` guarantees you

2. Bit-for-bit parity `fitz run` ↔ `fitz build`

4. `Result<T>` as error model

Lo que `smtp.send` te garantiza

2. Paridad bit-a-bit `fitz run` ↔ `fitz build`

4. `Result<T>` como modelo de error

`Secret<T>` redactado auto en logs

`@trace`/`@metric` solo sobre fns user, no sobre HTTP/WS

`Secret<T>` auto-redacted in logs

`@trace`/`@metric` only on user fns, not on HTTP/WS

OpenAPI con `bearerAuth` automático

`auth.blacklist` exige DbConn explícita

Mensaje de 403 enriquecido para `@requires`

`@requires` implica auth