DEV Community: Martin Palopoli

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

Cron jobs without Celery, Redis, or Beat: how Fitz puts a distributed scheduler inside the language

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

To run a task every 5 minutes in Python you need Celery + Redis + Celery Beat + a worker process + a dedicated Dockerfile. In Fitz it's a decorator. With retry, timezone, persistence, and catch-up inside the language.

The same story every time

Your client is happy with the API. It's running. Then they ask for three things:

"Can we clean up expired sessions every night?"
"Can the welcome email go out in the background so signup doesn't wait?"
"The daily report needs to run at 9 AM Buenos Aires time."

Three small asks. If you're in Python, the honest answer is: "OK, give me two days to add Celery."

The typical Python stack

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
        # heads up: if DST shifts, this fires at the wrong time
    },
}
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 via 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

Plus:

A supervisord.conf or systemd unit to keep things alive when they crash.
(Optional) flower for visibility — a 4th process.
Timezone conversions done by hand (Celery beat works in UTC, you have to compute the offset).
When a worker crashes between hitting the broker and finishing the task: does it retry? Does it hang? Depends on how you configured acks_late and task_reject_on_worker_lost.

Four processes in production. Three new libraries. A broker. New conventions. A day of setup.

The same thing in 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) {
    // expensive thing
}

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

docker-compose.yml:

services:
  api:
    build: .

That's it. One binary. One process. No broker. No dedicated worker. No beat. The scheduler runs inside the Fitz process.

The raw table

Item	Python (Celery + Redis + Beat)	Fitz
Initial setup	4 files + 3 services + 1 broker	1 decorator
Schedule	`crontab(hour=3, minute=0)` in config	`@cron("0 3 * * *")`
Timezone	UTC + manual offset	`tz="America/Argentina/Buenos_Aires"`
Retry/backoff	`autoretry_for=(...)` + `retry_backoff=True`	`retry={max=3, backoff="exponential", initial_secs=1, max_secs=30}`
Run persistence	Redis result backend + viewing via Flower	`store=db`, table auto-created in Postgres
Catch-up of missed runs	Not native	`catch_up=true`
Background fire-and-forget	`.delay(arg)` via broker	`spawn(fn_call)` direct
Type-checked args	None (everything JSON-serialized)	Static —`spawn` requires `@background` and refines to `Future<T>`
Production processes	api + worker + beat + redis	api
Docker images	3 (api, worker, beat) + redis	1

Opt-in persistence with `store=db`

When you want run history for auditing:

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

On boot, Fitz creates (if missing) fitz_cron_jobs and fitz_cron_runs. Each attempt is persisted with started_at/finished_at/status/attempt/error. You query with plain SQL:

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

No Flower install. No Sentry webhook. The DB you already have.

Catch-up

If the binary was down between 3 AM and 7 AM and the cron was at 3 AM:

Celery beat: the run is lost. The task won't fire again until the next midnight.
Fitz with catch_up=true: on boot, it sees there was a missed run between last_run_at and now, fires ONE immediate run (not N — avoids spam), then resumes the normal schedule.

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

Configurable retry with backoff

Three backoff modes: exponential (1s, 2s, 4s, 8s...), linear (1s, 2s, 3s, 4s...), constant (1s, 1s, 1s...). With 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 (capped at 60s if it would go higher)
}

In Python with 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():
    ...

Equivalent. But notice Celery requires you to enumerate the exceptions that trigger retry, and backoff is a boolean instead of a kind enum. Fitz retries on any error the fn returns (consistent with the language's Result model), and the backoff mode is explicit.

Real timezone, not hardcoded offsets

DST changes are the bug that wakes you up at 4 AM. Celery beat works in UTC and you have to remember that between March and November your 9 AM Buenos Aires cron runs at 12 UTC, but the rest of the year it shifts. Your client is in another timezone. Your app sells in three more countries.

Fitz accepts IANA timezones directly:

@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()         { ... }

Each runs at 9 AM local time in its tz, with DST handled by the underlying lib (chrono-tz). No manual conversions.

Background jobs without a broker

# Python
@app.post("/signup")
def signup(creds: Credentials):
    user = create_user(creds)
    send_welcome_email.delay(user.email)  # serialize 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))  // native tokio::spawn, typed
    return user
}

The compiler requires the fn be decorated with @background to authorize the spawn(...) — without it, the callsite throws a type error at build time. And it refines the return to Future<Null> with the concrete type of the target, not Any. If send_welcome_email returns Result<()>, the spawn(...) gives it to you typed.

When NOT to use @background and go back to Celery?

If you need jobs to survive process crashes → explicit persistence with store=db covers cron jobs; for @background with persistence, that's residual debt.
If you need to distribute jobs across N workers on N nodes → Celery with a shared broker is still the answer. @background runs in-process.

For 90% of services (nightly cleanup, transactional email, KPI recompute, cache sync) Fitz's model is enough and then some.

Cron-only mode for systemd

For services that ONLY have jobs (no HTTP), fitz build produces a binary that starts the scheduler and blocks on ctrl+c:

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

As a 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

Zero brokers. The binary is 5 MB. systemctl restart myapp-jobs and you're done.

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

This is what makes it shippable: the binary produced by fitz build runs the same scheduler as fitz run, with the same cron expression parser, the same retries, the same timezone. Same syntax, same semantics, no "oh, this only works in production if you configure Celery."

What Fitz does NOT give you (yet)

Being honest about where it doesn't reach:

Distributing jobs across N workers/nodes sharing a queue. @cron runs in-process. If you run two binaries with the same @cron, both fire — bug, not feature. For horizontal scaling of jobs, it's still Celery (or NATS JetStream, or Temporal). There's explicit debt in the roadmap about distributed locks.
A Flower-like UI. The data is in fitz_cron_jobs and fitz_cron_runs. If you want dashboards, external dashboards (Grafana, Metabase) cover it.
@background with persistence across restarts. For @cron it's in v0.11.2 (close of 9.w.3.iter2). For @background the spawn fires but doesn't survive crashes — deferred to iter3 if demand appears.
In-flight job cancellation. Today there's no API to "cancel all queued runs of X job". Process dies → in-flight runs die with it.

If you're in one of those cases, Fitz isn't the tool today. If not, the one-binary model covers the real case with fewer moving parts.

Closing

The argument for adding Celery to a Python API is typically: "but later it scales better." In 90% of services I've shipped over the past decade, that "later" never came — the app spent its entire lifetime with fewer than 100 jobs per hour and never needed a worker cluster.

For that 90%, Fitz replaces 4 processes + 3 libraries + 1 broker with one decorator. When you reach the other 10% where you need real horizontal scaling, Celery is still there — from python import celery is also available, you can do it yourself.

But starting the project with the simplest possible model and raising complexity only when you need it is the feedback loop Fitz wants to give you.

Next post in the series: "Auth with JWT, RBAC, and token blacklist without gluing 5 libraries: Fitz vs FastAPI + python-jose + passlib + Redis blacklist" — the full auth flow, side by side.

Repo: github.com/Thegreekman76/fitz
Chapter 30 of the guide (Jobs without Celery): docs/guide.md

Real-time chat with Fitz: typed WebSockets and AsyncAPI auto-generated

Martin Palopoli — Wed, 17 Jun 2026 12:33:06 +0000

Add a live click dashboard to the URL shortener from Part 2 using @ws("/path"), WsConn<T>, and the AsyncAPI 3.0 schema Fitz generates automatically. Same auth, same types, same binary.

The promise

In Part 2 we built a URL shortener: HTTP endpoints, Postgres ORM, JWT auth, native binary. Real and shippable, but every interaction is a request-response. Refresh the stats page to see the new click. Like 1998 again.

Today we make it live. When somebody clicks a short URL:

The HTTP handler still redirects and increments the counter (same as before).
A WebSocket subscribed to /dashboard gets a typed message with the click event.
The dashboard updates without polling.

The full diff against Part 2's code: ~40 lines. No new library installs. No websockets package, no socketio server, no redis for pub/sub. Just @ws on a function.

The typed WebSocket model

type ClickEvent {
    code: Str,
    target_url: Str,
    timestamp: Str,
}

@authenticated
@ws("/dashboard")
async fn dashboard(conn: WsConn<ClickEvent>, user: User) {
    log.info("dashboard.connected", { user_email: user.email })
    loop {
        let msg = match conn.recv() {
            Ok(m) => m,
            Err(_) => break,   // client disconnected
        }
        // For now we just echo back. In a real dashboard we'd ignore
        // incoming and only push out — `broadcast` is below.
        conn.send(msg)
    }
}

Three things in one decorator:

@ws("/dashboard") — register a WebSocket endpoint at that path.
WsConn<ClickEvent> — the typed connection. Every frame in or out is marshalled as ClickEvent.
@authenticated — auth runs before the WebSocket upgrade. Bad token → 401, no socket opened, no network resources wasted.

@ws understands the auth_provider from Part 2. The same JWT verification function gates the dashboard.

Auto-marshalling, both directions

The HTTP body deserialization from Part 1 — type-checked JSON, defaults applied, missing fields detected, extras rejected — also works for WebSocket frames.

When the dashboard sends a frame, Fitz serializes the ClickEvent to JSON and sends. When a frame comes in, Fitz deserializes the JSON into ClickEvent and validates. If a frame is malformed, the recv() returns Err.

You don't write a single json.dumps/json.loads call. The compiler did it.

In contrast: the typical Python WebSocket loop:

# typical FastAPI / websockets server
@app.websocket("/dashboard")
async def dashboard(websocket: WebSocket):
    await websocket.accept()
    while True:
        try:
            data = await websocket.receive_text()
            msg = ClickEvent.model_validate_json(data)  # pydantic
        except WebSocketDisconnect:
            break
        except ValidationError as e:
            await websocket.send_text(json.dumps({"error": str(e)}))
            continue
        await websocket.send_text(msg.model_dump_json())

In Fitz:

@ws("/dashboard")
async fn dashboard(conn: WsConn<ClickEvent>) {
    loop {
        let msg = match conn.recv() {
            Ok(m) => m,
            Err(_) => break,
        }
        conn.send(msg)
    }
}

The Python version has the same logic but you have to spell it. The Fitz version has the logic in the type.

Broadcasting clicks live

Now we wire the click event from the HTTP redirect to the dashboard. The trick: broadcast on a WsConn<T> sends to every connection on the same endpoint, not just the one that called it.

@get("/{code}")
async fn redirect(db: DbConn, code: Str) -> Result<HttpResponse> {
    let link: Link = match Link.where(fn(l) => l.code == code).first(db).await {
        Ok(l) => l,
        Err(_) => return Err("not found"),
    }
    // Same as Part 2: spawn a background increment.
    spawn(increment_clicks(db, link.id))
    // New: broadcast the click event to the dashboard.
    spawn(notify_dashboard(link.code, link.target_url))
    return Ok(redirect_to(link.target_url))
}

@background
async fn notify_dashboard(code: Str, target_url: Str) {
    let event = ClickEvent {
        code: code,
        target_url: target_url,
        timestamp: now_iso(),
    }
    // The runtime keeps the broadcaster for `/dashboard` reachable
    // from anywhere via the typed handle.
    ws.broadcast("/dashboard", event)
}

The redirect handler hasn't changed in shape — it still returns the redirect response. We added one spawn. The background fn notify_dashboard calls ws.broadcast which fans out to every connection currently subscribed to /dashboard.

The broadcast is fire-and-forget. If no dashboards are connected, the call is a no-op. If 50 dashboards are connected, all 50 get the event. The runtime handles the list of active connections.

Heartbeat, baked in

WebSocket connections silently die in production. Some proxy decides 60 seconds of idle is too long, drops the TCP connection, and your client thinks it's still connected. Every WebSocket library has to add heartbeats; in Fitz it's a flag on the server:

@server(43929, ws_heartbeat_secs=30)
fn main() => 0

Every 30 seconds, the runtime sends a Ping frame on every WebSocket connection. If the client doesn't respond with Pong, the runtime considers the connection dead and closes it cleanly. Most proxies, including Nginx and Cloudflare, accept this as "still alive" and don't drop it.

Set ws_heartbeat_secs=0 to disable (default is 30).

This is the kind of feature you'd never bother to add yourself in a small project, then spend a Sunday debugging when production breaks. It's defaulted on for the same reason tcp_keepalive is defaulted on.

AsyncAPI generated automatically

OpenAPI describes HTTP services. AsyncAPI is its event-driven sibling — same schema model, but for WebSockets, Kafka, MQTT, etc. Fitz generates AsyncAPI 3.0 automatically, the same way it generates OpenAPI for HTTP:

curl http://localhost:8080/asyncapi.json

{
  "asyncapi": "3.0.0",
  "info": { "title": "shortener", "version": "0.1.0" },
  "channels": {
    "/dashboard": {
      "messages": {
        "ClickEvent": {
          "payload": {
            "type": "object",
            "properties": {
              "code": { "type": "string" },
              "target_url": { "type": "string" },
              "timestamp": { "type": "string" }
            },
            "required": ["code", "target_url", "timestamp"]
          }
        }
      }
    }
  },
  "operations": {
    "/dashboard.receive": { ... },
    "/dashboard.send": { ... }
  },
  "components": {
    "securitySchemes": {
      "bearerAuth": { "type": "http", "scheme": "bearer", "bearerFormat": "JWT" }
    }
  }
}

This is the full event API of your service. I don't know another language that auto-generates AsyncAPI from typed source. The AsyncAPI ecosystem has client generators (TypeScript, Java, Python), which means: drop the schema into studio.asyncapi.com or asyncapi generate, get a typed client for your front-end.

If you mark a server with @server(docs=false), neither OpenAPI nor AsyncAPI is exposed. Defaults are on because the cost is small and the value is large.

A minimal browser client

A 30-line vanilla JS client to test the dashboard:

<!doctype html>
<input id="token" placeholder="paste JWT token">
<button id="connect">Connect</button>
<ul id="events"></ul>
<script>
document.getElementById("connect").onclick = () => {
    const token = document.getElementById("token").value
    const ws = new WebSocket("ws://localhost:8080/dashboard", [], {
        headers: { Authorization: `Bearer ${token}` }
    })
    // Note: browser WebSocket constructors don't accept custom headers
    // directly. In production, you'd put the token in a query param
    // (?token=...) and have the auth_provider read it from there.
    ws.onmessage = (e) => {
        const event = JSON.parse(e.data)
        const li = document.createElement("li")
        li.textContent = `${event.timestamp} • ${event.code} → ${event.target_url}`
        document.getElementById("events").prepend(li)
    }
    ws.onerror = (e) => console.error("ws error", e)
    ws.onclose = () => console.log("disconnected")
}
</script>

Put this in a file, open in the browser, paste a JWT, click the connect button. Then in another terminal:

# Get a token (from Part 2's POST /login)
TOKEN=$(curl -s localhost:8080/login -X POST -H 'content-type: application/json' \
    -d '{"email":"ada@example.com","password":"secret-ada-123"}' | jq -r .token)

# Create a short URL
curl -X POST localhost:8080/shorten -H "Authorization: Bearer $TOKEN" \
    -H 'content-type: application/json' \
    -d '{"target_url":"https://github.com/Thegreekman76/fitz"}'

# Click the short URL — the browser's `events` list updates instantly
curl -I localhost:8080/abc123

The dashboard <ul> populates with the click event the moment the redirect happens. The browser was holding open a WebSocket; the click event got broadcast; the JS callback rendered it. Live.

Limitations and trade-offs

Honestly:

One type per endpoint. WsConn<ClickEvent> means every frame in and out is ClickEvent. If you need both directions to be different types (In = ChatMessage, Out = ServerEvent), the workaround is to make the type a sum (union-style) — declare a wider Event type with optional fields. The cleaner solution (WsConn<In, Out> two-type generic) is on the roadmap.
No rooms / channels within an endpoint. broadcast goes to every connection on the endpoint. If you want "broadcast only to users subscribed to project 42", you maintain a Map<Int, Vec<WsConn>> yourself or split into multiple endpoints (one per project — works fine for low counts).
No reconnect with state replay. If the dashboard disconnects, on reconnect it sees new events only — there's no "give me the last 30 seconds I missed". Building that needs an event log (a Postgres table polled, or Redis Streams). Outside the WebSocket layer.
Binary frames are supported via WsConn<Bytes>, but I haven't talked about them here. Useful for file uploads or audio streaming; the AsyncAPI emits format: binary.

These are the honest gaps. The 90% of WebSocket use cases (real-time updates, chat, live dashboards, multi-user collaborative editing for a single document) are covered today.

How this composes with the rest of Fitz

You can mix WebSockets with everything else:

Auth: @authenticated/@admin/@requires("role") work on @ws, evaluated before the upgrade.
Middleware: middlewares run before the upgrade for things like rate limiting per IP.
ORM: the WebSocket handler can take a DbConn and query the database.
Async: recv/send/broadcast are awaitable; combine with HTTP calls or Postgres queries inside the loop.
Cron / spawn: cron jobs can ws.broadcast("/topic", event) to push periodic updates.
OpenTelemetry: the auth check before upgrade emits a trace span; subsequent broadcasts can be traced too.

Same language, same types, same binary. The WebSocket isn't a separate world.

What you'd need next for a real product

The dashboard above is enough to demo "real-time URL shortener clicks". For a production product you'd extend with:

Per-user filtering: only push clicks for URLs the user created. Trivial — check user.email == link.user_email before broadcasting, or split into one endpoint per user with the email in the path.
Aggregation: don't send every click; send a rate of clicks per code per second. Maintain state in the handler, send aggregated frames every 1 second.
Frontend framework integration: TypeScript types from the AsyncAPI schema. Run asyncapi generate once.

None of these change the Fitz code structure. They're all "edit the broadcast call site".

Try it

# Linux / macOS / WSL
curl -sSf https://thegreekman76.github.io/fitz/install.sh | sh

# Windows (PowerShell)
irm https://thegreekman76.github.io/fitz/install.ps1 | iex

# Reopen the terminal, then:
git clone https://github.com/Thegreekman76/fitz.git
cd fitz/boilerplates/api-websocket

# Read the README, run with docker compose or `fitz dev`
docker compose up

For VSCode (recommended — WsConn<T> hover, autocomplete on conn.send/recv/broadcast): grab the fitz-lang-<platform>.vsix from the releases page and code --install-extension fitz-lang-<platform>.vsix --force. The Language Server is bundled.

The api-websocket boilerplate is a typed chat server. Pretty much the same shape as the dashboard above — loop { recv; broadcast } with auth.

For the full URL shortener + live dashboard, the api-orm-full boilerplate has the whole thing assembled (HTTP routes + ORM + WebSocket dashboard + cron job + JWT auth) in ~250 lines total.

Repo: github.com/Thegreekman76/fitz
api-websocket boilerplate: github.com/Thegreekman76/fitz/tree/main/boilerplates/api-websocket
api-orm-full boilerplate (the full thing): github.com/Thegreekman76/fitz/tree/main/boilerplates/api-orm-full
Docs and course: thegreekman76.github.io/fitz
Guide chapter on WebSockets: thegreekman76.github.io/fitz/guide/#29-websockets
Roadmap: docs/roadmap.md
CHANGELOG: CHANGELOG.md
Issues: github.com/Thegreekman76/fitz/issues

If you build something with this, write to me. I want to see it.

Until the next one.

Chat en tiempo real con Fitz: WebSockets tipados y AsyncAPI auto-generado

Martin Palopoli — Wed, 17 Jun 2026 12:32:37 +0000

Sumá un dashboard de clicks en vivo al acortador de URLs de la Parte 2 usando @ws("/path"), WsConn<T> y el schema AsyncAPI 3.0 que Fitz genera automáticamente. Misma auth, mismos types, mismo binario.

La promesa

En la Parte 2 construimos un acortador de URLs: endpoints HTTP, ORM Postgres, auth JWT, binario nativo. Real y shippeable, pero cada interacción es request-response. Refrescá la página de stats para ver el click nuevo. Como 1998 otra vez.

Hoy lo hacemos vivo. Cuando alguien clickea un URL corto:

El handler HTTP sigue redirigiendo e incrementando el counter (igual que antes).
Un WebSocket suscrito a /dashboard recibe un mensaje tipado con el evento de click.
El dashboard se actualiza sin polling.

El diff completo contra el código de la Parte 2: ~40 líneas. Sin nuevos installs de librería. Sin paquete websockets, sin server socketio, sin redis para pub/sub. Solo @ws sobre una función.

El modelo de WebSocket tipado

type ClickEvent {
    code: Str,
    target_url: Str,
    timestamp: Str,
}

@authenticated
@ws("/dashboard")
async fn dashboard(conn: WsConn<ClickEvent>, user: User) {
    log.info("dashboard.connected", { user_email: user.email })
    loop {
        let msg = match conn.recv() {
            Ok(m) => m,
            Err(_) => break,   // cliente desconectó
        }
        // Por ahora solo hacemos echo. En un dashboard real ignoraríamos
        // los incoming y solo pusheríamos out — `broadcast` está abajo.
        conn.send(msg)
    }
}

Tres cosas en un decorador:

@ws("/dashboard") — registrar un endpoint WebSocket en ese path.
WsConn<ClickEvent> — la conexión tipada. Cada frame in o out se marshallea como ClickEvent.
@authenticated — la auth corre antes del upgrade WebSocket. Token malo → 401, sin socket abierto, sin recursos de red gastados.

@ws entiende el auth_provider de la Parte 2. La misma función de verificación de JWT gatea el dashboard.

Auto-marshalling, en ambas direcciones

La deserialización del body HTTP de la Parte 1 — JSON type-checked, defaults aplicados, fields faltantes detectados, extras rechazados — también funciona para frames WebSocket.

Cuando el dashboard envía un frame, Fitz serializa el ClickEvent a JSON y lo envía. Cuando un frame llega, Fitz deserializa el JSON en ClickEvent y valida. Si un frame está malformado, el recv() retorna Err.

No escribís ni una sola llamada a json.dumps/json.loads. El compilador lo hizo.

En contraste: el loop típico de WebSocket en Python:

# server típico de FastAPI / websockets
@app.websocket("/dashboard")
async def dashboard(websocket: WebSocket):
    await websocket.accept()
    while True:
        try:
            data = await websocket.receive_text()
            msg = ClickEvent.model_validate_json(data)  # pydantic
        except WebSocketDisconnect:
            break
        except ValidationError as e:
            await websocket.send_text(json.dumps({"error": str(e)}))
            continue
        await websocket.send_text(msg.model_dump_json())

En Fitz:

@ws("/dashboard")
async fn dashboard(conn: WsConn<ClickEvent>) {
    loop {
        let msg = match conn.recv() {
            Ok(m) => m,
            Err(_) => break,
        }
        conn.send(msg)
    }
}

La versión Python tiene la misma lógica pero tenés que deletrearla. La versión Fitz tiene la lógica en el type.

Broadcasteando clicks en vivo

Ahora cableamos el evento de click del redirect HTTP al dashboard. El truco: broadcast sobre un WsConn<T> envía a cada conexión sobre el mismo endpoint, no solo a la que lo llamó.

@get("/{code}")
async fn redirect(db: DbConn, code: Str) -> Result<HttpResponse> {
    let link: Link = match Link.where(fn(l) => l.code == code).first(db).await {
        Ok(l) => l,
        Err(_) => return Err("not found"),
    }
    // Igual que la Parte 2: spawnear un increment background.
    spawn(increment_clicks(db, link.id))
    // Nuevo: broadcastear el click event al dashboard.
    spawn(notify_dashboard(link.code, link.target_url))
    return Ok(redirect_to(link.target_url))
}

@background
async fn notify_dashboard(code: Str, target_url: Str) {
    let event = ClickEvent {
        code: code,
        target_url: target_url,
        timestamp: now_iso(),
    }
    // El runtime mantiene el broadcaster para `/dashboard` accesible
    // desde cualquier lugar vía el handle tipado.
    ws.broadcast("/dashboard", event)
}

El handler de redirect no cambió en shape — sigue retornando el response de redirect. Sumamos un spawn. La fn background notify_dashboard llama ws.broadcast que fanout a cada conexión actualmente suscrita a /dashboard.

El broadcast es fire-and-forget. Si no hay dashboards conectados, el call es no-op. Si hay 50 dashboards conectados, los 50 reciben el evento. El runtime maneja la lista de conexiones activas.

Heartbeat, horneado

Las conexiones WebSocket mueren silenciosamente en producción. Algún proxy decide que 60 segundos de idle es demasiado, droppea la conexión TCP, y tu cliente cree que sigue conectado. Cada librería WebSocket tiene que sumar heartbeats; en Fitz es una flag sobre el server:

@server(43929, ws_heartbeat_secs=30)
fn main() => 0

Cada 30 segundos, el runtime envía un frame Ping sobre cada conexión WebSocket. Si el cliente no responde con Pong, el runtime considera la conexión muerta y la cierra limpia. La mayoría de los proxies, incluyendo Nginx y Cloudflare, aceptan esto como "todavía vivo" y no la droppean.

Seteá ws_heartbeat_secs=0 para desactivar (default es 30).

Este es el tipo de feature que nunca te tomarías el trabajo de agregar en un proyecto chico, después pasarías un domingo debuggeando cuando se rompe en producción. Default on por la misma razón que tcp_keepalive está default on.

AsyncAPI generado automático

OpenAPI describe servicios HTTP. AsyncAPI es su hermano event-driven — mismo modelo de schema, pero para WebSockets, Kafka, MQTT, etc. Fitz genera AsyncAPI 3.0 automático, de la misma forma que genera OpenAPI para HTTP:

curl http://localhost:8080/asyncapi.json

{
  "asyncapi": "3.0.0",
  "info": { "title": "shortener", "version": "0.1.0" },
  "channels": {
    "/dashboard": {
      "messages": {
        "ClickEvent": {
          "payload": {
            "type": "object",
            "properties": {
              "code": { "type": "string" },
              "target_url": { "type": "string" },
              "timestamp": { "type": "string" }
            },
            "required": ["code", "target_url", "timestamp"]
          }
        }
      }
    }
  },
  "operations": {
    "/dashboard.receive": { ... },
    "/dashboard.send": { ... }
  },
  "components": {
    "securitySchemes": {
      "bearerAuth": { "type": "http", "scheme": "bearer", "bearerFormat": "JWT" }
    }
  }
}

Este es el API de eventos completo de tu servicio. No conozco otro lenguaje que auto-genere AsyncAPI desde código tipado. El ecosistema de AsyncAPI tiene generadores de clientes (TypeScript, Java, Python), lo que significa: dropeá el schema en studio.asyncapi.com o asyncapi generate, conseguí un cliente tipado para tu front-end.

Si marcás un server con @server(docs=false), ni OpenAPI ni AsyncAPI se exponen. Los defaults están on porque el costo es chico y el valor es grande.

Un cliente browser mínimo

Un cliente vanilla JS de 30 líneas para probar el dashboard:

<!doctype html>
<input id="token" placeholder="pegá JWT token">
<button id="connect">Conectar</button>
<ul id="events"></ul>
<script>
document.getElementById("connect").onclick = () => {
    const token = document.getElementById("token").value
    const ws = new WebSocket("ws://localhost:8080/dashboard", [], {
        headers: { Authorization: `Bearer ${token}` }
    })
    // Nota: los constructores de WebSocket del browser no aceptan headers
    // custom directamente. En producción, pondrías el token en un query
    // param (?token=...) y harías que el auth_provider lo lea de ahí.
    ws.onmessage = (e) => {
        const event = JSON.parse(e.data)
        const li = document.createElement("li")
        li.textContent = `${event.timestamp} • ${event.code} → ${event.target_url}`
        document.getElementById("events").prepend(li)
    }
    ws.onerror = (e) => console.error("ws error", e)
    ws.onclose = () => console.log("desconectado")
}
</script>

Poné esto en un archivo, abrí en el browser, pegá un JWT, clickeá el botón connect. Después en otra terminal:

# Conseguí un token (del POST /login de la Parte 2)
TOKEN=$(curl -s localhost:8080/login -X POST -H 'content-type: application/json' \
    -d '{"email":"ada@example.com","password":"secret-ada-123"}' | jq -r .token)

# Creá un URL corto
curl -X POST localhost:8080/shorten -H "Authorization: Bearer $TOKEN" \
    -H 'content-type: application/json' \
    -d '{"target_url":"https://github.com/Thegreekman76/fitz"}'

# Clickeá el URL corto — la lista `events` del browser se actualiza al instante
curl -I localhost:8080/abc123

El <ul> del dashboard se popula con el click event en el momento que el redirect ocurre. El browser tenía abierto un WebSocket; el click event se broadcasteó; el callback JS lo renderizó. En vivo.

Limitaciones y trade-offs

Honestamente:

Un tipo por endpoint. WsConn<ClickEvent> significa que cada frame in y out es ClickEvent. Si necesitás que ambas direcciones sean tipos distintos (In = ChatMessage, Out = ServerEvent), el workaround es hacer el tipo una suma (estilo union) — declarar un tipo Event más ancho con fields opcionales. La solución más limpia (WsConn<In, Out> generic de dos tipos) está en el roadmap.
Sin rooms / channels adentro de un endpoint. broadcast va a cada conexión sobre el endpoint. Si querés "broadcastear solo a users suscritos al proyecto 42", mantenés un Map<Int, Vec<WsConn>> vos o partís en múltiples endpoints (uno por proyecto — funciona bien para counts bajos).
Sin reconnect con replay de estado. Si el dashboard se desconecta, al reconectar ve solo eventos nuevos — no hay "dame los últimos 30 segundos que me perdí". Construir eso necesita un event log (una tabla Postgres poleada, o Redis Streams). Afuera de la capa WebSocket.
Frames binarios son soportados vía WsConn<Bytes>, pero no hablé de ellos acá. Útiles para uploads de archivos o audio streaming; el AsyncAPI emite format: binary.

Estos son los gaps honestos. El 90% de los casos de uso de WebSocket (updates en tiempo real, chat, dashboards en vivo, edición colaborativa multi-user para un solo documento) están cubiertos hoy.

Cómo esto compone con el resto de Fitz

Podés mezclar WebSockets con todo lo demás:

Auth: @authenticated/@admin/@requires("role") funcionan sobre @ws, evaluados antes del upgrade.
Middleware: middlewares corren antes del upgrade para cosas como rate limiting por IP.
ORM: el handler WebSocket puede tomar un DbConn y queryear la base de datos.
Async: recv/send/broadcast son awaiteables; combinalos con llamadas HTTP o queries Postgres adentro del loop.
Cron / spawn: jobs cron pueden ws.broadcast("/topic", event) para pushear updates periódicos.
OpenTelemetry: el chequeo de auth antes del upgrade emite un trace span; broadcasts subsecuentes también pueden trackearse.

Mismo lenguaje, mismos types, mismo binario. El WebSocket no es un mundo separado.

Qué necesitarías después para un producto real

El dashboard de arriba alcanza para demo "clicks de URL shortener en tiempo real". Para un producto en producción extenderías con:

Filtrado por usuario: solo pushear clicks para URLs que el user creó. Trivial — chequeá user.email == link.user_email antes de broadcastear, o partí en un endpoint por user con el email en el path.
Agregación: no enviar cada click; enviar una rate de clicks por código por segundo. Mantené estado en el handler, enviá frames agregados cada 1 segundo.
Integración con framework de frontend: tipos TypeScript desde el schema AsyncAPI. Corré asyncapi generate una vez.

Nada de esto cambia la estructura del código Fitz. Son todos "editá el call site del broadcast".

Probalo

# Linux / macOS / WSL
curl -sSf https://thegreekman76.github.io/fitz/install.sh | sh

# Windows (PowerShell)
irm https://thegreekman76.github.io/fitz/install.ps1 | iex

# Reabrí la terminal, después:
git clone https://github.com/Thegreekman76/fitz.git
cd fitz/boilerplates/api-websocket

# Leé el README, corré con docker compose o `fitz dev`
docker compose up

Para VSCode (recomendado — hover sobre WsConn<T>, autocomplete en conn.send/recv/broadcast): bajá el fitz-lang-<plataforma>.vsix desde la página de releases y code --install-extension fitz-lang-<plataforma>.vsix --force. El Language Server viene incluido.

El boilerplate api-websocket es un servidor de chat tipado. Más o menos el mismo shape que el dashboard de arriba — loop { recv; broadcast } con auth.

Para el shortener completo + dashboard en vivo, el boilerplate api-orm-full tiene la cosa entera ensamblada (rutas HTTP + ORM + dashboard WebSocket + cron job + auth JWT) en ~250 líneas total.

Repo: github.com/Thegreekman76/fitz
Boilerplate api-websocket: github.com/Thegreekman76/fitz/tree/main/boilerplates/api-websocket
Boilerplate api-orm-full (la cosa entera): github.com/Thegreekman76/fitz/tree/main/boilerplates/api-orm-full
Docs y curso: thegreekman76.github.io/fitz
Capítulo de la guía sobre WebSockets: thegreekman76.github.io/fitz/guide/#29-websockets
Roadmap: docs/roadmap.md
CHANGELOG: CHANGELOG.md
Issues: github.com/Thegreekman76/fitz/issues

Si construís algo con esto, escribime. Lo quiero ver.

Hasta la próxima.

Fitz CLI builder: como typer, pero en el lenguaje

Martin Palopoli — Sat, 13 Jun 2026 10:35:11 +0000

Construí herramientas CLI nativas en Fitz con @command, sin librería que instalar. Help auto-generado, flags con type coercion, positional args por convención, binario nativo a la salida. El mismo lenguaje que mueve servicios HTTP construye tus scripts.

¿Por qué un CLI builder en el lenguaje?

La mayoría del trabajo de CLI en Python vive en typer, click o argparse. Todas son librerías decentes; typer en particular es delicioso. La respuesta de Rust es clap. La de Go es cobra o urfave/cli. La de Node es commander.js o yargs. Cada lenguaje tiene una librería CLI. Cada una de ellas es una librería.

Una librería está bien hasta que te acordás que:

Las convenciones de la librería se imponen al resto de tu código (decoradores, factory objects, DSLs de builder).
El formato del help text es decisión de la librería, no del lenguaje.
Distribuir el resultado necesita un packager (pyinstaller, pyox, Docker, etc.) arriba.
El comportamiento cross-platform depende de la cobertura de la librería, no del lenguaje.
Agregar un --flag es cambiar la signature de una función más una llamada a decorador más quizás un objeto de config.

¿Qué pasaría si el mismo compilador que produce tu server HTTP también produce tus CLI tools, con el mismo type checker, el mismo async/await, el mismo Result<T> para errores, el mismo binario nativo como output?

Ese es el CLI builder de Fitz.

El hello-world completo

// main.fitz
@command("greet", desc="Saludar a una persona")
fn greet(name: Str, loud: Bool = false, count: Int = 1) -> Int {
    let n = count
    while n > 0 {
        if loud {
            print("HOLA, {name}!")
        } else {
            print("hola, {name}")
        }
        n = n - 1
    }
    return 0
}

Corrélo local durante el desarrollo:

$ fitz run main.fitz greet Ada
hola, Ada

$ fitz run main.fitz greet Ada --loud --count 3
HOLA, Ada!
HOLA, Ada!
HOLA, Ada!

Compilá a un binario self-contained:

$ fitz build
$ ls -lh ./greeter
-rwxr-xr-x  1  user  user   5.2M  Jun  5 14:00 ./greeter

$ ./greeter greet Ada --loud
HOLA, Ada!

Binario nativo de 5 MB, estáticamente linkeado excepto libc. Sin pyinstaller. Sin pyz. Sin intérprete Python en la máquina destino.

La convención: sin decoradores `@arg`/`@flag`

typer y click te piden que marques cada parámetro con un decorador que diga "esto es positional argument" o "esto es flag". Fitz usa una convención que cubre el mismo terreno sin la verbosidad:

Param sin default → positional argument requerido (mybin greet <name>).
Param con default → flag (--name <value>, o --loud para flags Bool).
Bool = false → flag bool (--loud lo activa, sin valor).
Int = N, Float = X, Str = "..." → flag con valor (--count 3).

La signature fn greet(name: Str, loud: Bool = false, count: Int = 1) -> Int le dice al compilador exactamente:

Param	Convención	CLI
`name: Str`	sin default → positional	`greet <name>`
`loud: Bool = false`	bool con default → flag	`--loud` (presencia = `true`)
`count: Int = 1`	int con default → flag con valor	`--count 3`

El trade-off: no podés tener positional optional args. Si querés uno, declaralo Str? (nullable) y hacé match en el body — el shape queda bien, el body maneja el caso missing explícito.

Conviví con este trade-off una semana antes de empezar a escribir CLIs en Fitz en lugar de typer, y no volví. La signature es la definición del CLI.

El help es auto-generado

$ ./mybin --help
USAGE:
    mybin <command> [ARGS] [OPTIONS]

COMMANDS:
    greet    Saludar a una persona

$ ./mybin greet --help
USAGE:
    mybin greet <name> [OPTIONS]

ARGUMENTS:
    <name>     (requerido)

OPTIONS:
    --loud, -l           default: false
    --count, -c <count>  default: 1
    -h, --help           mostrar este help

Short flags (-l, -c) se auto-derivan de la primera letra de la long flag. Si dos flags colisionarían en la misma letra, el compilador te avisa en build time (no en runtime, no cuando el user lo descubre):

✗ @command("greet") short flag conflict: `--loud` y `--limit`
  comparten primera letra `-l`. Renombrá una, u opt-out una con
  `@flag(short=null)`.

El formateo del output sigue convenciones de clap porque clap ya optimizó esto; no tiene sentido reinventarlo.

Multi-command dispatch

@command("greet", desc="Saludar a una persona")
fn greet(name: Str) -> Int { print("hola, {name}"); return 0 }

@command("add", desc="Sumar dos números")
fn add(a: Int, b: Int) -> Int { print("{a + b}"); return 0 }

@command("status", desc="Chequear estado del servicio")
async fn status(url: Str = "http://localhost:8080") -> Int {
    // ... llamada HTTP a la URL ...
    return 0
}

$ ./mybin --help
COMMANDS:
    greet     Saludar a una persona
    add       Sumar dos números
    status    Chequear estado del servicio

$ ./mybin add 21 21
42

$ ./mybin status --url http://prod.example.com
✓ healthy

El dispatch se genera en build time. No hay tabla de lookup de comandos en runtime que mantener.

Async nativo, porque por qué no

El comando status de arriba es async. El compilador lo detecta y envuelve el dispatch en #[tokio::main] automáticamente:

@command("fetch", desc="Traer una URL e imprimir el body")
async fn fetch(url: Str) -> Int {
    // Asumí que `http.get` fuera built-in (todavía no lo es; esto es ilustrativo).
    let body = http.get(url).await?
    print(body)
    return 0
}

El mismo async/await que usás en handlers HTTP funciona en CLI commands. Sin boilerplate de asyncio.run(). Sin "este comando es async, ese otro no, por favor no mezcles". Solo async fn cuando lo querés.

Exit codes

Las CLI tools viven y mueren por exit codes. La convención es POSIX:

0 — éxito.
1+ — retornado por el handler explícitamente.
2 — error de parsing del CLI (flag desconocida, type incorrecto, positional faltante).

@command("validate", desc="Chequear un archivo")
fn validate(path: Str) -> Int {
    let contents = read_file(path)
    if (contents.starts_with("FAIL")) {
        print("validación falló")
        return 1   // distinto de error interno
    }
    return 0
}

$ ./mybin validate ok.txt
$ echo $?
0

$ ./mybin validate fail.txt
validación falló
$ echo $?
1

$ ./mybin validate
✗ greet: missing positional argument `<path>`
$ echo $?
2

El parser CLI maneja los errores de parsing. Tu código maneja los errores de negocio. Limpio.

El poder completo del lenguaje

Acá es donde el approach de Fitz paga en una forma que typer no puede: tenés el lenguaje entero a tu disposición en CLI tools, no solo funciones y prints.

CLI tools con el ORM

¿Necesitás un CLI de admin de DB? Es el mismo ORM que el servicio HTTP:

@command("list-users", desc="Imprimir todos los users")
async fn list_users(db: DbConn) -> Int {
    let users = User.all(db).await
    for u in users {
        print("#{u.id}\t{u.email}\t{u.role}")
    }
    return 0
}

@command("delete-user", desc="Borrar un user por email")
async fn delete_user(db: DbConn, email: Str, force: Bool = false) -> Int {
    if not force {
        print("Usá --force para borrar de verdad")
        return 1
    }
    User.where(fn(u) => u.email == email).delete(db).await
    print("borrado")
    return 0
}

La inyección de DbConn funciona igual que en handlers HTTP. Closure-to-SQL en .where(...) funciona. Eager loading funciona. Las migraciones de fitz db diff/migrate funcionan. El CLI tool tiene el mismo acceso a tu capa de datos que el API server, con la misma seguridad de tipos.

CLI con llamadas HTTP

¿Querés un CLI que hable con tu propio API? Mismo lenguaje, sin SDK separado:

@command("ping-prod", desc="Pegarle al /healthz de prod")
async fn ping_prod() -> Int {
    let url = env_or("PROD_URL", "https://api.example.com")
    // ... usá la stdlib o `from python import requests` etc ...
    return 0
}

CLI como wrapper de herramientas Python

¿Necesitás envolver una herramienta Python pero con mejor UX? El interop Python está en Fitz:

from python import subprocess

@command("backup-db", desc="Correr pg_dump y subir a S3")
async fn backup_db(db_url: Str, bucket: Str) -> Int {
    let dump_cmd = "pg_dump {db_url} > backup.sql"
    subprocess.run(dump_cmd, shell=true).await?
    // ... upload a S3 ...
    return 0
}

El CLI está en Fitz. La llamada a pg_dump es shell. El upload a S3 podría ser Python con boto3 (from python import boto3). Un binario, tres ecosistemas compuestos.

CLI con config + secrets

El mismo Secret<T> y config(...) de la Parte 3 funcionan en CLI tools:

@command("deploy", desc="Empujar un release")
async fn deploy(env: Str = "staging") -> Int {
    let api_key: Secret<Str> = secret("DEPLOY_API_KEY")
    let endpoint: Str = config("DEPLOY_ENDPOINT", "https://deploy.internal")
    // ... usá api_key.expose() en el request ...
    return 0
}

api_key nunca leakea a logs incluso si el path de error del comando deploy imprime el config map. Las mismas reglas de redacción de los servicios HTTP aplican acá.

Boilerplate `cli-tool`

El repo tiene un boilerplate para proyectos CLI en boilerplates/cli-tool/. Trae tres comandos como starter:

report — genera un sumario de datos.
count — cuenta items.
regions — lista categorías.

fitz new my-cli --template cli-tool
cd my-cli
fitz dev

Hot reload también funciona en CLI mode — fitz dev watchea el source, mata y respawnea cuando guardás.

Lo que no está en la caja

Prompts interactivos (pensá inquirer / click.prompt). Fitz todavía no tiene readers de stdin estilo read_line en el core. Por ahora, aceptá una flag y alimentala vía env o stdin piped.
TUI (terminal UI con control de cursor). Fuera de scope.
Generación de shell completion (bash/zsh/fish). No se genera automático todavía. El approach del archivo _complete escrito a mano sigue funcionando.
--no-foo para flags Bool = true. Bools default a false; si querés una flag que default a true y vos opt-out, la convención es invertir el booleano en tu lógica de negocio por ahora. Trackeada como deuda menor.

Si algo de esto te bloquea, abrí un issue. O son chicos para agregar o ya están en progreso.

Por qué importa

Cada developer que construye servicios web también termina escribiendo CLI tools. Migraciones para correr, imports ad-hoc, smoke tests, scripts de deploy, acciones admin, exports de data. La elección hoy en la mayoría de los lenguajes es:

Mismo lenguaje que el servicio, con una librería CLI separada (Python: typer).
Bash / shell script que llama a tu servicio vía HTTP (perdés toda la seguridad de tipos).
Un lenguaje separado para ops (binarios Go al lado de un servicio Python — está bien, pero dos lenguajes para mantener).

Fitz propone una cuarta: mismo compilador, mismo type checker, mismo ORM, mismo async, mismo formato de binario. Tus CLI tools viven en el mismo repo, comparten el mismo type User, le pegan al mismo Postgres con la misma lógica de conexión.

Podés escribir mybin migrate y mybin reset-test-db en el mismo archivo .fitz que tus handlers @get("/users") si querés. O dividir en comandos separados por binario. La elección es tuya, no de la librería.

Probalo:

# Linux / macOS / WSL
curl -sSf https://thegreekman76.github.io/fitz/install.sh | sh

# Windows (PowerShell)
irm https://thegreekman76.github.io/fitz/install.ps1 | iex

# Reabrí la terminal, después:
fitz new micli --template cli-tool
cd micli
fitz run main.fitz greet --help
fitz build
./micli greet Ada --loud --count 3

Para VSCode (recomendado para editar — hover con tipos, autocomplete, signature help): bajá el fitz-lang-<plataforma>.vsix desde la página de releases y code --install-extension fitz-lang-<plataforma>.vsix --force. El Language Server viene incluido.

Vas a tener un binario nativo self-contained en tu pwd en menos de cinco minutos.

Repo: github.com/Thegreekman76/fitz
Boilerplate cli-tool: github.com/Thegreekman76/fitz/tree/main/boilerplates/cli-tool
Docs y curso: thegreekman76.github.io/fitz
Capítulo de la guía sobre CLI: thegreekman76.github.io/fitz/guide/#33-cli
Roadmap: docs/roadmap.md
CHANGELOG: CHANGELOG.md
Issues: github.com/Thegreekman76/fitz/issues

Hasta la próxima.

Fitz CLI builder: like typer, but in the language

Martin Palopoli — Sat, 13 Jun 2026 10:34:52 +0000

Build native CLI tools in Fitz with @command, no library to install. Help auto-generated, type-coerced flags, positional args by convention, native binary out the door. The same language that powers HTTP services builds your scripts.

Why a CLI builder in the language?

Most CLI work in Python lives in typer, click, or argparse. They're all decent libraries; typer in particular is delightful. The Rust answer is clap. The Go answer is cobra or urfave/cli. The Node answer is commander.js or yargs. Every language has a CLI library. Every one of them is a library.

A library is fine until you remember that:

The library's conventions are imposed on the rest of your code (decorators, factory objects, builder DSLs).
The help text formatting is the library's decision, not the language's.
Distributing the result needs a packager (pyinstaller, pyox, Docker, etc.) on top.
Cross-platform behavior depends on the library's coverage, not the language's.
Adding a --flag is changing a function signature plus a decorator call plus maybe a config object.

What if the same compiler that produces your HTTP server also produces your CLI tools, with the same type checker, the same async/await, the same Result<T> for errors, the same native binary as output?

That's the Fitz CLI builder.

The full hello-world

// main.fitz
@command("greet", desc="Greet a person")
fn greet(name: Str, loud: Bool = false, count: Int = 1) -> Int {
    let n = count
    while n > 0 {
        if loud {
            print("HELLO, {name}!")
        } else {
            print("hello, {name}")
        }
        n = n - 1
    }
    return 0
}

Run it locally during development:

$ fitz run main.fitz greet Ada
hello, Ada

$ fitz run main.fitz greet Ada --loud --count 3
HELLO, Ada!
HELLO, Ada!
HELLO, Ada!

Compile to a self-contained binary:

$ fitz build
$ ls -lh ./greeter
-rwxr-xr-x  1  user  user   5.2M  Jun  5 14:00 ./greeter

$ ./greeter greet Ada --loud
HELLO, Ada!

5 MB native binary, statically linked except libc. No pyinstaller. No pyz. No Python interpreter on the target machine.

The convention: no `@arg`/`@flag` decorators

typer and click ask you to mark every parameter with a decorator that says "this is a positional argument" or "this is a flag". Fitz uses a convention that covers the same ground without the verbosity:

Param without a default → positional argument required (mybin greet <name>).
Param with a default → flag (--name <value>, or --loud for Bool flags).
Bool = false → flag bool (--loud enables, no value).
Int = N, Float = X, Str = "..." → flag with value (--count 3).

The signature fn greet(name: Str, loud: Bool = false, count: Int = 1) -> Int tells the compiler exactly:

Param	Convention	CLI
`name: Str`	no default → positional	`greet <name>`
`loud: Bool = false`	bool with default → flag	`--loud` (presence = `true`)
`count: Int = 1`	int with default → flag with value	`--count 3`

The trade-off: you can't have positional optional args. If you want one, declare it Str? (nullable) and match on it in the body — the shape stays right, the body handles the missing case explicitly.

I lived with this trade-off for a week before I started writing CLIs in Fitz instead of typer, and I haven't gone back. The signature is the CLI definition.

Help is auto-generated

$ ./mybin --help
USAGE:
    mybin <command> [ARGS] [OPTIONS]

COMMANDS:
    greet    Greet a person

$ ./mybin greet --help
USAGE:
    mybin greet <name> [OPTIONS]

ARGUMENTS:
    <name>     (required)

OPTIONS:
    --loud, -l           default: false
    --count, -c <count>  default: 1
    -h, --help           show this help

Short flags (-l, -c) are auto-derived from the long flag's first letter. If two flags would collide on the same letter, the compiler tells you at build time (not at runtime, not when the user discovers it):

✗ @command("greet") short flag conflict: `--loud` and `--limit`
  share first letter `-l`. Rename one, or opt out one with
  `@flag(short=null)`.

The output formatting follows clap conventions because clap already optimized this; no point in reinventing.

Multi-command dispatch

@command("greet", desc="Greet a person")
fn greet(name: Str) -> Int { print("hello, {name}"); return 0 }

@command("add", desc="Sum two numbers")
fn add(a: Int, b: Int) -> Int { print("{a + b}"); return 0 }

@command("status", desc="Check service status")
async fn status(url: Str = "http://localhost:8080") -> Int {
    // ... HTTP call to the URL ...
    return 0
}

$ ./mybin --help
COMMANDS:
    greet     Greet a person
    add       Sum two numbers
    status    Check service status

$ ./mybin add 21 21
42

$ ./mybin status --url http://prod.example.com
✓ healthy

The dispatch is generated at build time. There's no runtime command lookup table to maintain.

Native async, because why not

The status command above is async. The compiler detects this and wraps the dispatch in #[tokio::main] automatically:

@command("fetch", desc="Fetch a URL and print the body")
async fn fetch(url: Str) -> Int {
    // Suppose `http.get` were a built-in (it's not yet; this is illustrative).
    let body = http.get(url).await?
    print(body)
    return 0
}

The same async/await you use in HTTP handlers works in CLI commands. No asyncio.run() boilerplate. No "this command is async, that one isn't, please don't mix". Just async fn when you want it.

Exit codes

CLI tools live and die by exit codes. The convention is POSIX:

0 — success.
1+ — returned by the handler explicitly.
2 — CLI parsing error (unknown flag, bad type, missing positional).

@command("validate", desc="Check a file")
fn validate(path: Str) -> Int {
    let contents = read_file(path)
    if (contents.starts_with("FAIL")) {
        print("validation failed")
        return 1   // distinct from internal error
    }
    return 0
}

$ ./mybin validate ok.txt
$ echo $?
0

$ ./mybin validate fail.txt
validation failed
$ echo $?
1

$ ./mybin validate
✗ greet: missing positional argument `<path>`
$ echo $?
2

The CLI parser handles the parsing errors. Your code handles the business errors. Clean.

The full power of the language

This is where the Fitz approach pays off in a way that typer can't: you have the entire language at your disposal in CLI tools, not just functions and prints.

CLI tools with the ORM

You need a database admin CLI? It's the same ORM as the HTTP service:

@command("list-users", desc="Print all users")
async fn list_users(db: DbConn) -> Int {
    let users = User.all(db).await
    for u in users {
        print("#{u.id}\t{u.email}\t{u.role}")
    }
    return 0
}

@command("delete-user", desc="Delete a user by email")
async fn delete_user(db: DbConn, email: Str, force: Bool = false) -> Int {
    if not force {
        print("Use --force to actually delete")
        return 1
    }
    User.where(fn(u) => u.email == email).delete(db).await
    print("deleted")
    return 0
}

DbConn injection works the same way it does in HTTP handlers. Closure-to-SQL in .where(...) works. Eager loading works. The migrations from fitz db diff/migrate work. The CLI tool has the same access to your data layer as the API server, with the same type safety.

CLI with HTTP calls

You want a CLI that talks to your own API? Same language, no separate SDK:

@command("ping-prod", desc="Hit the prod /healthz")
async fn ping_prod() -> Int {
    let url = env_or("PROD_URL", "https://api.example.com")
    // ... use the standard library or `from python import requests` etc ...
    return 0
}

CLI as a wrapper around Python tools

Need to wrap a Python tool but with better UX? Python interop is in Fitz:

from python import subprocess

@command("backup-db", desc="Run pg_dump and upload to S3")
async fn backup_db(db_url: Str, bucket: Str) -> Int {
    let dump_cmd = "pg_dump {db_url} > backup.sql"
    subprocess.run(dump_cmd, shell=true).await?
    // ... upload to S3 ...
    return 0
}

The CLI is in Fitz. The pg_dump call is shell. The S3 upload could be Python with boto3 (from python import boto3). One binary, three ecosystems composed.

CLI with config + secrets

The same Secret<T> and config(...) from Part 3 work in CLI tools:

@command("deploy", desc="Push a release")
async fn deploy(env: Str = "staging") -> Int {
    let api_key: Secret<Str> = secret("DEPLOY_API_KEY")
    let endpoint: Str = config("DEPLOY_ENDPOINT", "https://deploy.internal")
    // ... use api_key.expose() in the request ...
    return 0
}

api_key never leaks to logs even if the deploy command's error path prints the config map. The same redaction rules from HTTP services apply here.

Boilerplate `cli-tool`

The repo has a boilerplate for CLI projects at boilerplates/cli-tool/. It ships three commands as a starter:

report — generates a summary from data.
count — counts items.
regions — lists categories.

fitz new my-cli --template cli-tool
cd my-cli
fitz dev

Hot reload works in CLI mode too — fitz dev watches the source, kills and respawns when you save.

What's not in the box

Interactive prompts (think inquirer / click.prompt). Fitz doesn't have read_line style stdin readers in the core yet. For now, accept a flag and feed it via env or piped stdin.
TUI (terminal UI with cursor control). Not in scope.
Shell completion generation (bash/zsh/fish). Not generated automatically yet. The handwritten _complete file approach still works.
--no-foo for Bool = true flags. Bools default to false; if you want a flag that defaults to true and you opt out, the convention is to invert the boolean in your business logic for now. Tracked as a minor debt.

If any of these block you, file an issue. They're either small enough to add or already in progress.

Why this matters

Every developer who builds web services also ends up writing CLI tools. Migrations to run, ad-hoc imports, smoke tests, deploy scripts, admin actions, data exports. The choice today in most languages is:

Same language as the service, with a separate CLI library (Python: typer).
Bash / shell script that calls your service via HTTP (lose all type safety).
A separate language for ops (Go binaries next to a Python service — fine, but two languages to maintain).

Fitz proposes a fourth: same compiler, same type checker, same ORM, same async, same binary format. Your CLI tools live in the same repo, share the same type User, hit the same Postgres with the same connection logic.

You can write mybin migrate and mybin reset-test-db in the same .fitz file as your @get("/users") handlers if you want. Or split into separate commands per binary. The choice is yours, not the library's.

Try it:

# Linux / macOS / WSL
curl -sSf https://thegreekman76.github.io/fitz/install.sh | sh

# Windows (PowerShell)
irm https://thegreekman76.github.io/fitz/install.ps1 | iex

# Reopen the terminal, then:
fitz new mycli --template cli-tool
cd mycli
fitz run main.fitz greet --help
fitz build
./mycli greet Ada --loud --count 3

For VSCode (recommended for editing — hover with types, autocomplete, signature help): grab the fitz-lang-<platform>.vsix from the releases page and code --install-extension fitz-lang-<platform>.vsix --force. The Language Server is bundled.

You'll have a self-contained native binary in your pwd in under five minutes.

Repo: github.com/Thegreekman76/fitz
cli-tool boilerplate: github.com/Thegreekman76/fitz/tree/main/boilerplates/cli-tool
Docs and course: thegreekman76.github.io/fitz
Guide chapter on CLI: thegreekman76.github.io/fitz/guide/#33-cli
Roadmap: docs/roadmap.md
CHANGELOG: CHANGELOG.md
Issues: github.com/Thegreekman76/fitz/issues

Until the next one.

Del repo a producción con un solo comando: cómo Fitz hace del deployment una feature del lenguaje

Martin Palopoli — Thu, 11 Jun 2026 13:57:00 +0000

Un recorrido por la historia de deployment de Fitz — healthchecks, secrets como tipos opacos, observability con OpenTelemetry, Dockerfiles autogenerados y fitz deploy. Production-ready no es una checklist, es sintaxis.

La historia de deployment que la mayoría de los lenguajes no cuenta

El primer 80% de un servicio es divertido: rutas, types, lógica de negocio, tests. El último 20% es la parte que efectivamente entrega la cosa, y ahí es donde todo el mundo pega con cinta cinco herramientas distintas:

Una librería de healthcheck estilo psutil porque Kubernetes quiere /healthz.
python-decouple o pydantic-settings para env vars, más tu propia clase Secret que con suerte no termina en logs.
opentelemetry-instrumentation-fastapi más opentelemetry-exporter-otlp-proto-http más opentelemetry-instrumentation-sqlalchemy más cualquiera sea la combinación correcta de versiones este mes.
Un Dockerfile copiado de un post de blog, con tres cosas mal para tu setup.
Un docker-compose.yml que está bien excepto por el env var que tiene que venir de .env.production.
Un Makefile o justfile con los comandos reales, o un YAML de CI que hace el equivalente.

Llevo diez años haciendo esto en cada proyecto. Es la parte donde el lenguaje deja de ayudarte. Fitz se niega a hacer eso. Deployment está en el lenguaje.

Acá está el stack completo de producción en Fitz, y qué reemplaza cada pieza.

Health checks como decoradores

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

@healthz
fn liveness() -> Bool => true

@readyz
async fn readiness(db: DbConn) -> Bool {
    return match db.exec("SELECT 1").await {
        Ok(_) => true,
        Err(_) => false,
    }
}

/healthz y /readyz se auto-montan en el router HTTP. Nada para importar, ninguna librería para configurar. El return value de la función determina el HTTP status (200 si true, 503 si false). Kubernetes contento.

El compilador también hace cumplir el shape: la función retorna Bool o Result<Bool>, toma un DbConn si lo necesita (el runtime lo inyecta). Si te olvidás del @readyz entero, el endpoint de readiness simplemente no existe — no hay librería que "casi configures" mal.

En Python:

# requirements.txt: starlette-healthcheck, pyhealthcheck, ...
from healthcheck import HealthCheck
health = HealthCheck()
def db_check():
    try:
        db.execute("SELECT 1")
        return True, "ok"
    except Exception as e:
        return False, str(e)
health.add_check(db_check)
app.add_route("/healthz", health.run)  # acordate de montarlo

En Fitz:

@readyz
async fn readiness(db: DbConn) -> Bool {
    return match db.exec("SELECT 1").await { Ok(_) => true, Err(_) => false }
}

Eso es todo. Misma semántica, sin librería, sin paso de mounting, sin docs que tenés que volver a leer.

Secrets como tipos opacos

El bug que más vi en código de producción: alguien loguea la password, la API key, el JWT secret. El logger entrega a Loki / Splunk / Sentry / lo que sea. El secret está ahora en logs de producción. Todo el mundo está de acuerdo en que esto es malo. Nadie tiene una buena respuesta en las librerías estándar — os.environ["FOO"] es solo un string. pydantic.SecretStr existe pero tenés que opt-in en todos lados y la gente se olvida.

Fitz hace de los secrets un tipo de primera clase:

let JWT_SECRET: Secret<Str> = secret("JWT_SECRET")
let DB_URL: Secret<Str> = secret("DATABASE_URL")
let LOG_LEVEL: Str = config("LOG_LEVEL", "info")

Tres cosas siguen del hecho de que Secret<T> sea un tipo:

print(JWT_SECRET) imprime `""`*. El trait Display redacta el valor.
La serialización JSON redacta. log.info("config", { jwt: JWT_SECRET }) envía "jwt": "***" a tu sistema de observability.
Hay exactamente una forma de exponer el valor: JWT_SECRET.expose(). Explícita, grepeable. El code review puede auditar cada call site en segundos.

config("LOG_LEVEL", "info") es el hermano no-secret. Mismo lookup de env var, mismo default, tipo Str plano — sin redacción porque no es sensible. El sistema de tipos hace la distinción obvia en lugar de depender de convenciones de nombres.

Combinado con el migrator (Parte 2), los secrets de Postgres viven en Secret<Str> desde el momento en que entran al binario hasta que se entregan al driver. No hay variable string con la password dando vueltas.

Observability con un solo env var

Tracing y métricas son las partes de observability de producción que todo el mundo quiere y nadie quiere configurar. El SDK Python de OpenTelemetry actualmente te pide que:

Instales opentelemetry-api, opentelemetry-sdk, el exporter para tu protocolo, y un paquete de instrumentation por cada librería que uses.
Configures el tracer provider, el meter provider, los resource attributes, el sampler.
Lo enganches en un hook de startup.
Esperes que ninguna de las versiones de esos paquetes se haya roto entre releases de Python.

En Fitz:

OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318 ./mybin

Esa es la integración. Cada request HTTP abre un span que exporta a OTLP. El span lleva http.method, http.target (template de la ruta, no path con params), http.status_code, duration_ms. El trace_id y span_id se propagan a cada log.info(...) adentro del handler — cuando grepeás Jaeger por trace_id, encontrás cada log line relacionado en todos los servicios al instante.

Cuando el env var no está seteado: cero overhead, cero llamadas de red, ninguna tarea de exporter corriendo.

Podés opt-out por ruta:

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

Podés agregar spans explícitos adentro de la lógica de negocio:

@trace(name="process_order")
@metric(name="orders")
async fn process(order: Order) -> Result<Receipt> {
    let validated = validate(order)?
    let charged = charge(validated).await?
    return Ok(receipt_for(charged))
}

@trace abre un tracing::info_span!. @metric registra un histograma <name>_duration_seconds y un counter <name>_calls_total, populados al hacer drop del scope de la función — funciona con paths return explícitos, sin código muerto.

Las métricas también se exponen en /metrics (formato Prometheus) si lo activás:

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

Para tiendas OpenTelemetry-native, el exporter OTLP envía también las métricas — ambos backends funcionan.

Logs estructurados con auto-correlación

log.info("user.signup", {
    user_id: user.id,
    email: user.email,
    plan: "free",
})

Eso emite una línea JSON a stderr con timestamp, level, msg, los fields explícitos, y automáticamente el trace_id/span_id del request HTTP actual. Sin llamada a tracer.get_current_span(). El wrapper HTTP setea el contexto; el logger lo lee. Correlacionás logs y traces por trace_id sin pensar en eso.

Si valores Secret<T> aparecen en los kwargs, se redactan antes de serializar. No hay forma de loguear accidentalmente un secret salvo escribir .expose() explícito.

Pretty-print en dev, JSON en producción:

$ ./mybin
2026-06-05 14:23:11 INFO http.access method=GET target=/users/{id} status=200 duration_ms=12 trace_id=ab12cd34...
2026-06-05 14:23:11 INFO user.lookup user_id=42 found=true

$ FITZ_LOG_FORMAT=json ./mybin
{"timestamp":"2026-06-05T14:23:11Z","level":"INFO","msg":"http.access","method":"GET","target":"/users/{id}","status":200,"duration_ms":12,"trace_id":"ab12cd34..."}

El happy path de Loki/Datadog/Splunk es el mismo JSON estés usando OpenTelemetry o no. Ningún agente re-parsea prefijos.

Feature flags como decorador

Los feature flags estilo unleash/launchdarkly son normalmente una llamada de servicio por evaluación. Para la mayoría de los proyectos, la respuesta correcta es mucho más simple: una flag en tu config, un override por env var, un 404 si la flag está off.

@flag("new-checkout")
@post("/v2/checkout")
fn v2_checkout(body: Cart) -> Receipt { ... }

Dos fuentes:

Sección [flags] en fitz.toml — defaults compile-time horneados al binario.
Env vars FITZ_FLAG_<NAME> — override runtime sin recompilar.

Default es false (fail-safe). Cuando la flag está off, el handler HTTP/WS retorna 404 — y la ruta está gateada antes de que corran los middlewares y la auth, así que no gastás ciclos en algo que el user no puede alcanzar de todas formas.

Adentro de la lógica de negocio:

if flag("show-experimental-banner") {
    show_banner()
}

flags.list() enumera flags conocidas (config + env). flags.is_enabled("name") es el alias.

El modelo no está tratando de reemplazar a LaunchDarkly. Está tratando de eliminar la excusa para commitear if user.id == 42 y gatear features para testing. Servicio de feature flags real cuando necesitás targeting, porcentajes, audit log. Flags built-in cuando solo querés un kill switch.

Dockerfile generado de tu AST

fitz docker init

Lee tu main.fitz y escribe tres archivos:

Dockerfile — multi-stage. El builder usa la imagen toolchain de Fitz. La stage de runtime es gcr.io/distroless/cc-debian12 (o python:3.12-slim-bookworm si hay un from python import ... en tu código — distroless no puede hostear CPython).
.dockerignore — defaults que matchean los smell tests (target/, .git/, .env*, __pycache__/).
docker-compose.yml — tu app más la infraestructura que necesita. db.connect(...) en tu código agrega postgres:16-alpine con healthcheck y volumen pgdata. @server(8080) setea EXPOSE 8080. @cron agrega restart: unless-stopped. Healthcheck contra /healthz porque hay un decorador @healthz.

La detección es AST-only (~50ms). No ejecuta tu programa, no toca el disco, no probe ports. Lee el árbol sintáctico, busca decoradores y calls landmarks, llena el template correcto.

Esta es la parte de deployment que tuve mal en cada proyecto: el Dockerfile que está casi bien pero no tiene libpq para Postgres, el compose que está casi bien pero montea el volumen equivocado. fitz docker init lo agarra bien la primera vez porque el AST le dice qué necesitás.

Los archivos se commitean. Editalos cuando los necesites:

$ fitz docker init        # generar
$ # editás Dockerfile, editás docker-compose.yml — son archivos normales
$ git add Dockerfile docker-compose.yml .dockerignore
$ git commit

fitz docker build es el wrapper fino que corre docker build -t <pkg-name>:latest . con el working directory correcto.

fitz deploy

# Build de imagen y push al registry (saltá el push con --no-push).
fitz deploy docker --tag mycorp/api:v1

# Levantá el stack compose local (con -d por default; --no-detach para foreground).
fitz deploy compose

Estos no son herramientas nuevas. Son wrappers finos sobre docker build/docker push y docker compose up. El punto no es inventar un sistema de deploy; el punto es que el comando de deploy existe en la misma toolchain que fitz build. No tenés que mantener un deploy.sh al lado de tu código.

Targets en el MVP: docker y compose. Targets explícitamente fuera del MVP: fly, railway, k8s. Para esos, corré flyctl deploy / railway up / kubectl apply directo — ya son buenas herramientas, Fitz no necesita re-envolverlas. Si aparece demanda por un target específico más adelante, se puede agregar — el helper crate son ~430 líneas.

Cómo se ve end-to-end

Un servicio real en Fitz hoy:

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

let DB_URL: Secret<Str> = secret("DATABASE_URL")
let JWT_SECRET: Secret<Str> = secret("JWT_SECRET")
let db = db.connect(DB_URL.expose())

@healthz
fn liveness() -> Bool => true

@readyz
async fn readiness() -> Bool {
    return match db.exec("SELECT 1").await { Ok(_) => true, Err(_) => false }
}

@table("users")
type User { @primary id: Int, email: Str, name: Str, role: Str }

@auth_provider
fn check_token(headers: Map<Str, Str>) -> Result<User> { /* verify JWT */ }

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

@trace(name="charge")
@metric(name="charges")
@requires("billing")
@post("/charge")
async fn charge(body: ChargeRequest, user: User) -> Result<Receipt> {
    log.info("charge.attempt", { user_id: user.id, amount: body.amount })
    let receipt = stripe_charge(body).await?
    return Ok(receipt)
}

@cron("0 0 3 * * *", retry={max: 5, backoff: "exponential"}, store=db)
async fn cleanup_expired_tokens() {
    auth.cleanup_expired(db).await?
}

Lo deployás:

fitz docker init    # genera Dockerfile + compose con postgres + healthcheck
git add . && git commit -m "deploy setup"
fitz deploy docker --tag mycorp/api:v1

En producción:

docker run --rm \
  -e DATABASE_URL=postgres://... \
  -e JWT_SECRET=... \
  -e OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4318 \
  -p 8080:8080 \
  mycorp/api:v1

Ese binario:

Auto-montea /healthz, /readyz, /openapi.json, /docs, /metrics, /asyncapi.json.
Exporta cada request HTTP como span OpenTelemetry con trace_id propagado a logs.
Valida JWTs con passwords hasheadas Argon2id.
Corre cleanup scheduled con retry persistente sobre las tablas fitz_cron_jobs/fitz_cron_runs.
Redacta cada Secret<T> de logs, JSON responses, y fields estructurados.
Maneja SIGTERM gracefully (drainando requests, después exit).
Compila a ~18 MB de binario nativo.

Escribiste ~50 líneas de código de negocio. El resto es el lenguaje haciendo lo que el lenguaje debería hacer.

Lo que todavía no está acá (honesto)

Te dije la verdad en la Parte 1: un solo dev. La historia de deployment tiene gaps conocidos:

fitz deploy fly / fitz deploy railway / fitz deploy k8s no están construidos. Usá flyctl deploy o railway up o kubectl apply directo. Los CLIs nativos son excelentes — Fitz no necesita re-envolverlos hoy.
Config de sidecar log shipping en el compose generado. Si querés Fluent Bit / Vector / Loki Promtail cableado, editá el compose a mano. El autogen cubre el shape del programa, no el backend de observability.
Limits de CPU/memoria en compose. El MVP no los setea — deploys de producción (compose stack a un server) deberían agregar deploy.resources.limits.
SBOM / firma de imagen. No se autogenera. La imagen es output de docker build. Firmá con cosign si lo necesitás.

Todo lo demás de arriba está en v0.15.0 hoy, con tests, con paridad bit-a-bit fitz run ↔ fitz build, con ejemplos en los docs.

Por qué importa

El split 80/20 que describí al principio — 80% código divertido, 20% pegoteo de producción — es un impuesto sobre cada lenguaje diseñado antes de 2015. Los lenguajes que diseñaron para producción desde el día uno (Go es el ejemplo obvio) cambiaron otras cosas para llegar ahí. Fitz está tratando de mantener el feel gradual-typed, expression-rich, async-first de Python — y aún así taxar producción con cero peso extra.

Si pasaste por una semana de deploy y sentiste "esto no debería requerir cinco pestañas de Stack Overflow", ese es el sentimiento al que estoy construyendo para eliminar.

Probalo:

# Linux / macOS / WSL
curl -sSf https://thegreekman76.github.io/fitz/install.sh | sh

# Windows (PowerShell)
irm https://thegreekman76.github.io/fitz/install.ps1 | iex

# Reabrí la terminal, después:
fitz new mi-api-prod --http
cd mi-api-prod
# Editá main.fitz para sumar @healthz, @table, una ruta HTTP
fitz docker init
fitz deploy compose

Para VSCode (recomendado — hover con tipos, autocomplete, signature help): bajá el fitz-lang-<plataforma>.vsix desde la página de releases y code --install-extension fitz-lang-<plataforma>.vsix --force. El Language Server viene incluido.

Vas a tener un servicio con healthchecks, observability, y Docker compose en tu proyecto en menos de cinco minutos. Avisame qué se rompió.

Repo: github.com/Thegreekman76/fitz
Docs y curso: thegreekman76.github.io/fitz
Capítulo de la guía sobre deployment: thegreekman76.github.io/fitz/guide/#35-deployment
Roadmap: docs/roadmap.md
CHANGELOG: CHANGELOG.md
Issues: github.com/Thegreekman76/fitz/issues

Hasta la próxima.

DEV Community: Martin Palopoli

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

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

Lo mismo en Fitz

La tabla cruda

Por partes

Spans HTTP auto-instrumentados

Trace_id propagado a logs custom

Métricas con un decorador

Endpoint Prometheus opcional

Export OTLP con un env var

Secret<T> redactado auto en logs

Decisiones de diseño que vale la pena entender

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

Acceso log auto-emitido

Storage del span context con task-local

Lo que Fitz NO te da (todavía)

Cierre

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

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

The same thing in Fitz

The raw table

Piece by piece

Auto-instrumented HTTP spans

Trace_id propagated to custom logs

Metrics with one decorator

Optional Prometheus endpoint

OTLP export with one env var

Secret<T> auto-redacted in logs

Design decisions worth understanding

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

Auto-emitted access log

Span context storage with task-local

What Fitz does NOT give you (yet)

Closing

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

TL;DR

Qué hace fitzwatch

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

La propuesta

Cómo sumarte

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

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

El stack típico de Python

Lo mismo en Fitz

La tabla cruda

Por partes

Hash de passwords

JWT

RBAC custom apilable

Token blacklist persistente

OpenAPI con bearerAuth automático

El checker hace gran parte del trabajo

Decisiones de diseño que vale la pena entender

auth.blacklist exige DbConn explícita

Mensaje de 403 enriquecido para @requires

@requires implica auth

Lo que Fitz NO te da (todavía)

Cierre

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

Real auth needs more pieces than you'd think

The typical Python stack

The same thing in Fitz

The raw table

Piece by piece

Password hashing

JWT

Stackable custom RBAC

Persistent token blacklist

Automatic bearerAuth in OpenAPI

The checker does much of the work

Design decisions worth understanding

auth.blacklist requires an explicit DbConn

Enriched 403 message for @requires

@requires implies auth

What Fitz does NOT give you (yet)

Closing

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

La historia de siempre

El stack típico de Python

`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

Automatic `bearerAuth` in OpenAPI

`auth.blacklist` requires an explicit DbConn

Enriched 403 message for `@requires`

`@requires` implies auth

Persistencia opt-in con `store=db`

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

Opt-in persistence with `store=db`

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

La convención: sin decoradores `@arg`/`@flag`

Boilerplate `cli-tool`

The convention: no `@arg`/`@flag` decorators