Robyn por dentro: o framework Python com runtime Rust, do primeiro endpoint ao benchmark com FastAPI

Frameworks web em Python costumam acelerar o desenvolvimento, mas quase sempre entregam a parte mais quente do servidor a outra camada: Gunicorn, Uvicorn, Hypercorn ou algum servidor WSGI/ASGI. O Robyn muda essa divisão. Você continua escrevendo rotas em Python, mas o runtime HTTP, o roteamento e parte da coordenação das requisições vivem em Rust.

Essa combinação chama atenção porque promete o melhor de dois mundos: a produtividade do Python na regra de negócio e um núcleo nativo para reduzir o custo de receber, encaminhar e devolver cada requisição. A promessa é atraente, mas não basta repetir que “Rust é rápido”. Precisamos entender onde o Rust realmente participa, onde o Python continua sendo o limite e quanto dessa vantagem sobrevive quando adicionamos validação, banco de dados, autenticação e I/O.

Neste guia, vamos construir uma API progressivamente, explorar rotas, parâmetros, Pydantic, middlewares, injeção de dependências, autenticação, OpenAPI, WebSockets, SSE, streaming, testes, múltiplos processos e integração direta com Rust. No fim, reproduzimos um benchmark entre Robyn 0.88.0 e FastAPI 0.139.2, com o código completo, percentis de latência e uma leitura honesta dos resultados.

A ideia central: Robyn pode reduzir bastante o overhead do caminho HTTP, mas nenhum runtime transforma automaticamente código Python bloqueante, consultas lentas ou arquitetura ruim em uma aplicação rápida.

1 O que é Robyn e por que ele existe

Robyn é um framework web open source para CPython cujo runtime é implementado em Rust. A API pública continua sendo Python: criamos uma instância de Robyn, registramos handlers com decorators e retornamos strings, dicionários ou objetos de resposta. Por baixo, a camada nativa recebe conexões, encontra a rota e coordena a execução.

Isso o diferencia de FastAPI e Starlette, que seguem o padrão ASGI e normalmente rodam sobre Uvicorn ou outro servidor compatível. ASGI oferece interoperabilidade: framework, servidor e middleware podem ser trocados dentro do mesmo contrato. Robyn escolhe integração vertical. Seu servidor e framework evoluem juntos, o que abre espaço para otimizações, mas reduz a portabilidade para componentes ASGI.

Não se trata de “reescrever a aplicação em Rust”. O código de domínio, os clientes de banco, as bibliotecas e os handlers continuam em Python. O Rust cuida de uma parte importante da infraestrutura, não substitui todo o programa.

1.1 O que o Rust acelera — e o que continua em Python

O ganho potencial aparece principalmente no trabalho repetido em cada requisição: aceitar conexões, analisar HTTP, localizar rotas, transportar dados entre o runtime e o handler, agendar trabalho e montar a resposta. Em endpoints muito pequenos, esse overhead representa grande parte do tempo total; por isso benchmarks de “Hello World” tendem a favorecer bastante arquiteturas nativas.

Quando a rota executa uma query de 80 ms, chama três APIs externas ou serializa milhares de objetos, o custo do framework vira apenas uma fração da latência. Robyn não acelera automaticamente:

• SQL lento ou sem índice;
• cliente HTTP síncrono dentro de async def;
• loops e cálculos escritos em Python;
• contenção em Redis, filas ou locks;
• serialização e validação desnecessariamente complexas;
• limites de rede, proxy ou banco.

A pergunta útil não é “Rust é mais rápido que Python?”, e sim “quanto do tempo desta requisição pertence ao runtime e quanto pertence ao meu sistema?”.

1.2 Versão, maturidade e compatibilidade

Este artigo foi validado com Robyn 0.88.0, publicado em junho de 2026, e Python 3.14.6. O pacote declara suporte a Python 3.10 ou superior e distribui wheels para plataformas e versões específicas do CPython. Como existe extensão nativa, vale confirmar a disponibilidade de wheel para seu sistema antes de escolher uma imagem base exótica ou uma arquitetura menos comum.

O projeto está em desenvolvimento ativo. Isso é positivo para recursos e correções, mas também exige disciplina: fixe versões, acompanhe o changelog e teste upgrades. A série 0.x não deve ser tratada com a mesma expectativa de estabilidade de uma API que promete compatibilidade semântica consolidada.

2 A arquitetura Python + Rust por dentro

Quando o processo inicia, as funções Python decoradas são registradas no runtime. O roteador nativo armazena método, caminho, tipo do handler, middlewares e metadados. Quando uma requisição chega, a parte Rust resolve a rota e cria os objetos que serão expostos ao Python. O handler executa e o resultado atravessa novamente a fronteira para virar uma resposta HTTP.

2.1 O ciclo de uma requisição

1. O cliente abre ou reutiliza uma conexão e envia HTTP.
2. O runtime Rust analisa método, URL, headers e corpo.
3. O roteador encontra o handler e os middlewares aplicáveis.
4. Robyn prepara os parâmetros injetados e chama Python.
5. O handler executa código síncrono ou uma coroutine.
6. O retorno é convertido para texto, bytes, JSON ou streaming.
7. Middlewares posteriores podem alterar status e headers.
8. O runtime envia a resposta e decide se mantém a conexão aberta.

Cruzar Python e Rust tem custo. A vantagem aparece quando o trabalho economizado no runtime é maior do que a sobrecarga da fronteira FFI. Por isso mover uma função trivial para Rust pode não compensar, enquanto um algoritmo pesado e autocontido pode se beneficiar muito.

2.2 Sync, async e GIL sem confusão

Robyn aceita def e async def. Handlers assíncronos são indicados quando a rota espera por recursos que oferecem APIs não bloqueantes, como banco assíncrono, HTTP ou filas. Handlers síncronos são úteis para bibliotecas bloqueantes e fluxos simples.

import asyncio
from robyn import Robyn

app = Robyn(__file__)

@app.get("/sync")
def sync_handler():
    return {"mode": "sync"}

@app.get("/async")
async def async_handler():
    await asyncio.sleep(0.01)
    return {"mode": "async"}

O runtime Rust não remove o GIL do bytecode Python. Duas rotas executando cálculo Python puro continuam competindo pelas regras do interpretador. O que o Rust consegue fazer sem manter o GIL é trabalho nativo de rede e coordenação; quando entra no handler Python, as restrições do CPython voltam a importar.

3 Instalação e primeiro projeto

Comece em um ambiente virtual e fixe a versão usada pelo projeto:

python -m venv .venv
source .venv/bin/activate
python -m pip install "robyn[pydantic]==0.88.0"

Para incluir também recursos opcionais como templates Jinja2, use robyn[all]. O primeiro arquivo pode ser pequeno:

from robyn import Robyn

app = Robyn(__file__)

@app.get("/")
async def home():
    return {"message": "Olá, Robyn!"}

if __name__ == "__main__":
    app.start(host="0.0.0.0", port=8080)

Execute com python app.py e abra http://localhost:8080. Por padrão, a documentação Swagger fica em /docs e o documento OpenAPI em /openapi.json.

3.1 CLI, desenvolvimento e configuração

O mesmo arquivo aceita opções do Robyn:

python app.py --dev
python app.py --workers 4 --processes 2
python app.py --log-level INFO
python app.py --disable-openapi
python app.py --fast

--dev ativa recarregamento durante o desenvolvimento. Não use reload em produção. Workers e processos precisam ser dimensionados por teste; dobrar os números pode aumentar contenção, memória e conexões com o banco em vez de melhorar throughput.

Variáveis específicas podem ser colocadas em robyn.env, como limite de payload. Configuração de negócio e segredos devem seguir uma camada própria, carregada de ambiente ou cofre de segredos, sem confundir configuração do runtime com configuração da aplicação.

4 Rotas, requests e responses

Os principais métodos HTTP possuem decorators próprios. Parâmetros dinâmicos usam :nome, segmentos opcionais recebem ? e caminhos catch-all usam *nome.

@app.get("/products")
async def list_products(): ...

@app.post("/products")
async def create_product(): ...

@app.get("/products/:product_id")
async def get_product(product_id: int):
    return {"id": product_id}

@app.get("/files/*path")
async def nested_file(request):
    return {"path": request.path_params["path"]}

Além de GET, POST, PUT, PATCH e DELETE, a API expõe HEAD, OPTIONS, CONNECT e TRACE. Isso não significa que todos devam ser habilitados indiscriminadamente; publique apenas os métodos que fazem parte do contrato.

4.1 Path, query, headers e body

Para casos comuns, Robyn resolve parâmetros tipados diretamente da assinatura. No exemplo abaixo, product_id vem do path, page e active vêm da query string:

from robyn import Request

@app.get("/products/:product_id")
async def product_detail(
    request: Request,
    product_id: int,
    page: int = 1,
    active: bool = True,
):
    trace_id = request.headers.get("x-trace-id")
    return {
        "id": product_id,
        "page": page,
        "active": active,
        "trace_id": trace_id,
    }

Para acesso completo, Request reúne query_params, path_params, headers, body, method, url, form_data, files, ip_addr e a identidade autenticada. request.json() interpreta o corpo preservando strings, números, booleanos, listas, objetos e null.

@app.post("/inspect")
async def inspect(request: Request):
    payload = request.json()
    return {
        "payload": payload,
        "content_type": request.headers.get("content-type"),
        "client_ip": request.ip_addr,
    }

4.2 JSON, status, headers e cookies

Strings viram texto; dicionários e listas são serializados para JSON. Quando precisamos controlar o protocolo, usamos Response:

from robyn import Headers, Response, status_codes

@app.post("/orders")
async def create_order():
    response = Response(
        status_code=status_codes.HTTP_201_CREATED,
        headers=Headers({
            "Content-Type": "application/json",
            "Location": "/orders/42",
        }),
        body='{"id":42,"status":"created"}',
    )
    response.set_cookie(
        key="session",
        value="opaque-token",
        http_only=True,
        secure=True,
        same_site="Strict",
        max_age=3600,
    )
    return response

Em produção, prefira tokens imprevisíveis, assine ou armazene a sessão no servidor e nunca coloque informação sensível em cookie legível pelo cliente.

5 Validação com Pydantic e OpenAPI

Comparativos antigos costumavam apontar validação manual como uma diferença decisiva para FastAPI. No Robyn 0.88, Pydantic v2 pode ser ativado por anotação no handler. A validação é opt-in: rotas sem um modelo não pagam esse custo.

from pydantic import BaseModel, Field
from robyn import Robyn

app = Robyn(__file__)

class ProductIn(BaseModel):
    name: str = Field(min_length=3, max_length=120)
    price: float = Field(gt=0)
    quantity: int = Field(ge=1, le=10_000)
    tags: list[str] = Field(default_factory=list)

class ProductOut(ProductIn):
    id: int

@app.post("/products")
async def create_product(product: ProductIn) -> ProductOut:
    return ProductOut(id=42, **product.model_dump())

Robyn encontra o BaseModel pela anotação, interpreta o JSON, valida e injeta a instância. O tipo de retorno também alimenta o schema OpenAPI e pode ser serializado automaticamente.

5.1 Erros 422 e contratos de resposta

Um preço inválido retorna 422 com a localização, o tipo do erro, a mensagem e o valor recebido. Isso aproxima bastante a experiência de FastAPI, embora detalhes do envelope e da localização possam ser diferentes. Clientes não devem depender da mensagem humana completa; normalize um contrato de erro estável para sua aplicação.

{
  "error": "Validation Error",
  "detail": [
    {
      "type": "float_parsing",
      "loc": ["price"],
      "msg": "Input should be a valid number",
      "input": "invalid"
    }
  ]
}

A documentação automática é disponibilizada em /docs. É possível fornecer configuração OpenAPI customizada ao construtor, declarar tags e marcar rotas autenticadas. Em ambientes internos, avalie desabilitar a interface pública ou protegê-la; esconder Swagger, sozinho, não protege a API.

6 SubRouters e organização de projeto

Uma prova de conceito cabe em app.py; uma aplicação real não deveria concentrar rotas, domínio e infraestrutura no mesmo arquivo. SubRouter permite prefixos e módulos por contexto:

# routes/products.py
from robyn import SubRouter

products = SubRouter(prefix="/products")

@products.get("")
async def list_products():
    return {"items": []}

@products.get("/:product_id")
async def get_product(product_id: int):
    return {"id": product_id}

# app.py
from robyn import Robyn
from routes.products import products

app = Robyn(__file__)
api = SubRouter(prefix="/api/v1")
api.include_router(products)
app.include_router(api)

Os prefixos de routers aninhados são acumulados. Uma organização sustentável separa transporte HTTP, casos de uso e acesso a dados. Assim, trocar framework ou testar domínio não exige simular uma requisição para tudo.

7 Middlewares e injeção de dependências

Hooks before_request executam antes do handler; after_request podem alterar a resposta. Eles podem ser globais ou associados a uma rota, e versões atuais aceitam cadeias de múltiplos middlewares.

from time import perf_counter

@app.before_request()
async def start_timer(request):
    request.headers.set("x-started-at", str(perf_counter()))
    return request

@app.after_request()
async def add_timing(request, response):
    started = float(request.headers.get("x-started-at"))
    elapsed_ms = (perf_counter() - started) * 1000
    response.headers.set("server-timing", f"app;dur={elapsed_ms:.2f}")
    return response

Use middleware para preocupações transversais: correlação, autenticação, métricas, CORS e headers de segurança. Evite esconder regra de negócio na cadeia; ordem, retornos antecipados e tratamento de exceção ficam difíceis de entender.

A injeção nativa possui dois escopos. inject_global disponibiliza dependências globais, enquanto inject registra dependências no router. Os parâmetros reservados precisam se chamar global_dependencies e router_dependencies.

class ProductService:
    async def list(self):
        return []

app.inject_global(product_service=ProductService())

@app.get("/products")
async def list_products(global_dependencies):
    service = global_dependencies["product_service"]
    return {"items": await service.list()}

7.1 Estado compartilhado e múltiplos processos

Uma dependência em memória não é automaticamente compartilhada entre processos. Se você iniciar quatro processos, pode haver quatro pools, quatro caches e quatro contadores. Conexões de banco devem ser abertas no ciclo de vida correto de cada processo; estado que precisa ser global deve morar em PostgreSQL, Redis ou outro serviço apropriado.

Também não trate um dicionário Python injetado como cache distribuído. Ele pode funcionar em desenvolvimento e produzir inconsistência quando a aplicação escala horizontalmente.

8 Autenticação, autorização, CORS e sessões

Robyn fornece AuthenticationHandler, extração de Bearer token e Identity. A rota opta por autenticação com auth_required=True:

from robyn import Request
from robyn.authentication import (
    AuthenticationHandler,
    BearerGetter,
    Identity,
)

class TokenAuth(AuthenticationHandler):
    async def authenticate(self, request: Request):
        token = self.token_getter.get_token(request)
        claims = await verify_token(token)
        if claims is None:
            return None
        return Identity(claims=claims)

app.configure_authentication(
    TokenAuth(token_getter=BearerGetter())
)

@app.get("/me", auth_required=True)
async def me(request: Request):
    return request.identity.claims

Autenticar responde “quem é?”. Autorizar responde “pode fazer isto neste recurso?”. Não pare no token: verifique escopo, tenant, propriedade e ação dentro do caso de uso. Um usuário autenticado ainda pode tentar acessar o pedido de outra pessoa.

CORS é habilitado com ALLOW_CORS(app, origins=[...]). Liste origens exatas em produção e lembre que CORS é uma política do navegador, não uma barreira contra curl, bots ou comunicação servidor a servidor.

Para sessões, defina cookies HttpOnly, Secure e SameSite, faça rotação de identificador no login e expire o estado no logout. Nunca confie em um identificador fornecido pelo cliente sem validação e assinatura.

9 Exceções, logs e observabilidade

Um handler global pode converter exceções inesperadas em resposta consistente:

import logging
from robyn import Response

logger = logging.getLogger("api")

@app.exception
async def handle_exception(error: Exception):
    logger.exception("request_failed", exc_info=error)
    return Response(
        status_code=500,
        headers={"Content-Type": "application/json"},
        body='{"error":"internal_error"}',
    )

Não devolva stack trace, SQL ou segredo ao cliente. Registre o detalhe internamente com request ID. Em produção, o mínimo observável inclui taxa de requisições, latência p50/p95/p99, status, exceções, saturação dos workers, uso de CPU/memória, conexões de banco e tempo das dependências externas.

A fronteira Python–Rust acrescenta uma dimensão ao diagnóstico. Um erro pode nascer no handler, na conversão de tipos ou no runtime nativo. Fixar versão e manter símbolos/logs adequados reduz o custo de investigar problemas raros.

10 WebSockets, SSE e streaming

Robyn oferece comunicação bidirecional por WebSocket e streaming HTTP. Escolha pelo fluxo: WebSocket quando cliente e servidor conversam continuamente; SSE quando apenas o servidor envia eventos; StreamingResponse para bytes, relatórios e arquivos produzidos em partes.

10.1 WebSockets bidirecionais

from robyn import WebSocketDisconnect

@app.websocket("/chat")
async def chat(websocket):
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_json({"echo": message})
    except WebSocketDisconnect:
        pass

@chat.on_connect
async def connected(websocket):
    await websocket.send_text("connected")

@chat.on_close
async def disconnected(websocket):
    return None

Esse primeiro exemplo confirma o fluxo: receive_text() suspende a coroutine até chegar uma mensagem, send_json() responde apenas ao cliente atual e WebSocketDisconnect encerra o loop sem transformar uma desconexão normal em erro da aplicação.

10.1.1 Exemplo completo: chat em tempo real

Agora podemos montar uma sala pública funcional. O cliente informa o nome na query string, envia JSON e recebe broadcasts de todos que estão conectados ao mesmo endpoint. O servidor limita o tamanho e ignora formatos inválidos:

import json
from datetime import UTC, datetime

from robyn import Robyn, WebSocketDisconnect

app = Robyn(__file__)

@app.websocket("/ws/chat")
async def chat_room(websocket, name: str = "anonymous"):
    safe_name = name.strip()[:40] or "anonymous"

    await websocket.broadcast(json.dumps({
        "type": "presence",
        "message": f"{safe_name} entrou na conversa",
    }))

    try:
        while True:
            raw_message = await websocket.receive_text()
            try:
                payload = json.loads(raw_message)
            except json.JSONDecodeError:
                await websocket.send_json({
                    "type": "error",
                    "message": "Envie um objeto JSON válido",
                })
                continue

            if not isinstance(payload, dict):
                await websocket.send_json({
                    "type": "error",
                    "message": "O JSON precisa ser um objeto",
                })
                continue

            message = str(payload.get("message", "")).strip()

            if not message:
                await websocket.send_json({
                    "type": "error",
                    "message": "A mensagem não pode estar vazia",
                })
                continue

            if len(message) > 500:
                await websocket.send_json({
                    "type": "error",
                    "message": "Limite de 500 caracteres",
                })
                continue

            await websocket.broadcast(json.dumps({
                "type": "message",
                "author": safe_name,
                "message": message,
                "sent_at": datetime.now(UTC).isoformat(),
            }))
    except WebSocketDisconnect:
        # Desconectar é parte normal do ciclo de vida do socket.
        pass

@chat_room.on_connect
async def chat_connected(websocket, name: str = "anonymous"):
    return json.dumps({
        "type": "connected",
        "client_id": websocket.id,
        "name": name[:40],
    })

@chat_room.on_close
async def chat_disconnected(websocket, name: str = "anonymous"):
    return None

if __name__ == "__main__":
    app.start(host="0.0.0.0", port=8080)

O navegador pode se conectar sem biblioteca adicional. Salve o arquivo abaixo como chat.html, abra-o e informe um nome:

<!doctype html>
<html lang="pt-BR">
<head>
  <meta charset="utf-8">
  <title>Chat com Robyn</title>
</head>
<body>
  <input id="name" value="Paulo" placeholder="Seu nome">
  <button id="connect">Conectar</button>
  <ul id="messages"></ul>
  <form id="form">
    <input id="message" maxlength="500" autocomplete="off">
    <button>Enviar</button>
  </form>

  <script>
    let socket;
    const list = document.querySelector("#messages");

    document.querySelector("#connect").onclick = () => {
      const name = encodeURIComponent(
        document.querySelector("#name").value
      );
      socket = new WebSocket(
        `ws://localhost:8080/ws/chat?name=${name}`
      );

      socket.onmessage = (event) => {
        const data = JSON.parse(event.data);
        const item = document.createElement("li");
        item.textContent = data.author
          ? `${data.author}: ${data.message}`
          : data.message || `Conectado como ${data.name}`;
        list.appendChild(item);
      };

      socket.onclose = () => {
        const item = document.createElement("li");
        item.textContent = "Conexão encerrada";
        list.appendChild(item);
      };
    };

    document.querySelector("#form").onsubmit = (event) => {
      event.preventDefault();
      const input = document.querySelector("#message");

      if (socket?.readyState === WebSocket.OPEN && input.value.trim()) {
        socket.send(JSON.stringify({message: input.value}));
        input.value = "";
      }
    };
  </script>
</body>
</html>

Abra o arquivo em duas abas para observar o broadcast. Em produção, troque ws:// por wss://, autentique a conexão, valide Origin, limite mensagens por cliente, implemente heartbeat e defina o que acontece quando o consumidor fica lento. O nome da query string é apenas didático: identidade real deve vir de credencial verificada.

O broadcast mostrado é local ao endpoint atendido pela instância. Com múltiplos processos, containers ou servidores, clientes podem cair em instâncias diferentes. Redis Pub/Sub, NATS ou Kafka distribuem os eventos; um armazenamento compartilhado mantém presença e histórico. Sticky session pode ajudar a manter a conexão, mas não substitui o barramento quando uma mensagem precisa alcançar todas as réplicas.

10.2 SSE e downloads em streaming

import asyncio
from robyn import SSEMessage, SSEResponse, StreamingResponse

@app.get("/events")
async def events(request):
    async def generate():
        for index in range(5):
            await asyncio.sleep(1)
            yield SSEMessage(
                data=f"progress:{index}",
                event="progress",
                id=str(index),
            )
    return SSEResponse(generate())

@app.get("/export")
async def export(request):
    async def rows():
        async for row in repository.iter_rows():
            yield f"{row.id},{row.name}\n".encode()
    return StreamingResponse(rows(), media_type="text/csv")

Streaming evita carregar todo o arquivo na memória, mas não elimina pressão. Se o cliente lê devagar, o produtor precisa respeitar backpressure. Defina timeouts compatíveis no proxy e encerre recursos quando houver desconexão.

11 Uploads, arquivos estáticos e templates

O objeto Request expõe form_data e files. O exemplo abaixo valida nome e tamanho antes de persistir:

from pathlib import Path
from uuid import uuid4

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/uploads")
async def upload(request):
    content = request.files.get("file")
    if content is None:
        return {"error": "file_required"}
    if len(content) > 5 * 1024 * 1024:
        return {"error": "file_too_large"}

    name = f"{uuid4().hex}.bin"
    (UPLOAD_DIR / name).write_bytes(content)
    return {"file": name}

Em produção, limite payload no servidor e proxy, detecte o tipo real, não confie na extensão, armazene fora da árvore pública e considere antivírus. write_bytes é bloqueante; para arquivos grandes, use armazenamento em objeto e upload direto ou desloque a gravação para uma thread.

Robyn também serve diretórios estáticos, arquivos e templates Jinja2. Isso é conveniente para painéis pequenos. Para assets pesados, CDN ou proxy dedicado geralmente oferece cache, compressão e range requests mais maduros.

12 Testes automatizados

TestClient executa o pipeline em processo, sem abrir porta. Ele cobre rotas, parâmetros, middlewares e respostas com testes rápidos:

from robyn.testing import TestClient
from app import app

def test_create_product():
    with TestClient(app) as client:
        response = client.post(
            "/products",
            json_data={
                "name": "Teclado",
                "price": 299.90,
                "quantity": 2,
            },
        )

    assert response.status_code == 200
    assert response.json()["name"] == "Teclado"

def test_rejects_invalid_price():
    client = TestClient(app)
    response = client.post(
        "/products",
        json_data={"name": "X", "price": -1, "quantity": 0},
    )
    assert response.status_code == 422

O cliente em processo não substitui testes de integração com o servidor real. Mantenha uma camada pequena que exercite socket, proxy, streaming, limites de payload, shutdown e comportamento multiprocess. O benchmark também deve usar rede local real, não TestClient.

13 Escalabilidade, workers e respostas constantes

Robyn expõe --workers e --processes. Workers aumentam concorrência dentro da arquitetura do runtime; processos permitem usar múltiplos núcleos e isolam o GIL entre interpretadores. A combinação ideal depende do workload e da memória.

python app.py --workers 4 --processes 2

Comece com poucos processos, meça CPU e p99 e aumente gradualmente. Calcule o impacto no banco: oito processos com pool de vinte conexões podem abrir até 160 conexões antes de considerar jobs e deploys simultâneos.

Rotas constantes podem ser pré-computadas com const=True:

@app.get("/health/live", const=True)
async def liveness():
    return {"status": "ok"}

Use apenas quando a resposta realmente não depende da requisição, do horário, do usuário ou de estado externo. Liveness pode ser constante; readiness normalmente precisa verificar se a instância consegue atender e não deveria mentir sobre dependências essenciais.

13.1 Quando levar uma função para Rust

A CLI consegue criar e compilar módulos Rust que depois são importados como módulos Python via PyO3/rustimport. Isso faz sentido para algoritmos CPU-bound, parsing pesado, compressão especializada ou transformação de grandes buffers. Não faz sentido reescrever uma função que passa a maior parte do tempo esperando PostgreSQL.

use pyo3::prelude::*;

#[pyfunction]
fn sum_squares(values: Vec<i64>) -> i64 {
    values.into_iter().map(|value| value * value).sum()
}

#[pymodule]
fn native_math(module: &Bound<'_, PyModule>) -> PyResult<()> {
    module.add_function(wrap_pyfunction!(sum_squares, module)?)?;
    Ok(())
}

python app.py --create-rust-file native_math
python app.py --compile-rust-path ./native

Meça incluindo conversão de dados. Copiar uma lista enorme de objetos Python para Rust pode consumir o ganho. Prefira interfaces estreitas, buffers ou tipos simples, trate panics e produza wheels no CI para as plataformas suportadas.

13.2 Ciclo de vida, timeouts e recursos avançados

Recursos externos devem acompanhar o ciclo de vida da aplicação. Abra pools, clientes HTTP e conexões no evento de startup de cada processo e feche tudo no shutdown. Isso evita criar conexões no momento da importação, antes do fork, ou abandonar sockets durante deploy.

import httpx

state = {}

@app.startup_handler
async def startup():
    state["http"] = httpx.AsyncClient(
        timeout=httpx.Timeout(5.0),
        limits=httpx.Limits(max_connections=100),
    )

@app.shutdown_handler
async def shutdown():
    await state["http"].aclose()

Além do timeout do cliente externo, o próprio servidor aceita client_timeout e keep_alive_timeout. Na versão analisada, os padrões documentados são 30 e 20 segundos. Valores menores não são automaticamente melhores: uploads, streaming e clientes móveis exigem orçamento diferente.

app.start(
    host="0.0.0.0",
    port=8080,
    client_timeout=30,
    keep_alive_timeout=20,
)

Robyn também documenta suporte experimental a io_uring em Linux. Experimental significa medir, testar kernel e manter fallback; não habilite apenas porque o nome sugere mais desempenho.

13.2.1 GraphQL, plugins, MCP e IA

O suporte GraphQL é construído por integração com Strawberry e a própria documentação o classifica como estágio inicial. Na prática, uma rota recebe a query, variáveis e nome da operação e delega a execução ao schema Strawberry. Antes de adotar, valide autenticação por resolver, limite de profundidade e complexidade, batching e observabilidade.

O ecossistema de plugins inclui integrações como rate limiting, mas ainda é menor do que o universo ASGI. Audite manutenção, compatibilidade e ordem dos middlewares de cada plugin; limite de requisição distribuído precisa de armazenamento compartilhado, não apenas memória local.

Versões atuais também anunciam rotas para agentes de IA e servidor MCP. Esses recursos podem facilitar a exposição de ferramentas, mas não alteram as regras de segurança: valide schema, autentique clientes, aplique autorização por ferramenta, limite custo e tempo, registre chamadas e trate prompt ou resposta de modelo como entrada não confiável. Para uma API tradicional, você pode simplesmente não instalar nem usar essas capacidades.

13.2.2 Deploy com Docker

Quando a aplicação usa apenas o wheel publicado do Robyn, a imagem não precisa carregar o toolchain Rust. Um Dockerfile mínimo fica assim:

FROM python:3.14-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

USER 10001
EXPOSE 8080

CMD ["python", "app.py", "--log-level", "INFO", "--workers", "2", "--processes", "1"]

Se houver módulo Rust próprio, compile em um estágio separado e copie apenas o artefato para a imagem final. Fixe a versão da base, gere SBOM, verifique vulnerabilidades, execute como usuário sem privilégios e deixe TLS, compressão e limites adicionais no proxy ou ingress quando fizer sentido.

14 Benchmark reproduzível: Robyn vs FastAPI

Os artigos usados como ponto de partida relatam vantagem expressiva do Robyn, mas usam máquinas, versões e cargas diferentes. Alguns misturam endpoint mínimo, Fibonacci em Python e implementação nativa em Rust. Esses são experimentos válidos para perguntas distintas, porém não devem ser combinados em uma única frase como se medissem a mesma coisa.

Para este artigo, preparei três rotas equivalentes:

• JSON mínimo: mede principalmente runtime, roteamento e serialização;
• validação: recebe JSON, valida com o mesmo Pydantic e calcula um campo;
• I/O simulado: espera 10 ms sem bloquear, mostrando o que acontece quando a espera domina.

14.1 Ambiente e metodologia

Logs de acesso foram desativados. Os handlers equivalentes eram assíncronos para evitar que o FastAPI enviasse uma função síncrona ao thread pool enquanto o Robyn tomasse outro caminho. Cada framework rodou em porta separada e o mesmo wrk gerou a carga.

14.2 Código usado no benchmark

Aplicação Robyn:

import asyncio
from pydantic import BaseModel
from robyn import Robyn

app = Robyn(__file__)

class Item(BaseModel):
    name: str
    price: float
    quantity: int
    active: bool = True

@app.get("/json")
async def json_response():
    return {
        "message": "ok",
        "framework": "python",
        "items": [1, 2, 3],
    }

@app.post("/validate")
async def validate(item: Item):
    return {
        "name": item.name,
        "total": item.price * item.quantity,
        "active": item.active,
    }

@app.get("/io")
async def simulated_io():
    await asyncio.sleep(0.01)
    return {"message": "ok"}

if __name__ == "__main__":
    app.start(host="127.0.0.1", port=8081)

Aplicação FastAPI:

import asyncio
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    quantity: int
    active: bool = True

@app.get("/json")
async def json_response():
    return {
        "message": "ok",
        "framework": "python",
        "items": [1, 2, 3],
    }

@app.post("/validate")
async def validate(item: Item):
    return {
        "name": item.name,
        "total": item.price * item.quantity,
        "active": item.active,
    }

@app.get("/io")
async def simulated_io():
    await asyncio.sleep(0.01)
    return {"message": "ok"}

Comandos dos servidores:

python robyn_app.py --log-level ERROR --workers 1 --processes 1

uvicorn fastapi_app:app \
  --host 127.0.0.1 \
  --port 8082 \
  --workers 1 \
  --loop uvloop \
  --http h11 \
  --no-access-log \
  --log-level error

Script Lua para o POST:

wrk.method = "POST"
wrk.body = '{"name":"notebook","price":19.9,"quantity":3,"active":true}'
wrk.headers["Content-Type"] = "application/json"

wrk -t4 -c100 -d15s --latency http://127.0.0.1:8081/json
wrk -t4 -c100 -d15s --latency http://127.0.0.1:8082/json

wrk -t4 -c100 -d15s --latency -s post.lua http://127.0.0.1:8081/validate
wrk -t4 -c100 -d15s --latency -s post.lua http://127.0.0.1:8082/validate

14.3 Resultados e interpretação

No endpoint mínimo, Robyn atingiu aproximadamente 2,53 vezes o throughput do FastAPI nesta configuração. Com Pydantic, a relação ficou em cerca de 2,17 vezes. Quando adicionamos 10 ms de espera assíncrona, caiu para aproximadamente 1,16 vez.

Essa queda não é uma derrota; é a interpretação correta. No primeiro teste, quase todo o orçamento pertence ao framework e ao servidor. No terceiro, a espera deliberada ocupa a maior parte do tempo. Quanto mais banco, rede e regra de negócio entram na rota, menor tende a ser a participação relativa do runtime.

O p99 também merece atenção. Robyn apresentou cauda menor nos três testes locais. Ainda assim, uma rodada curta em notebook não caracteriza estabilidade de produção. Para capacidade real, execute múltiplas repetições, descarte outliers explicáveis, monitore CPU e memória e teste durante minutos ou horas.

14.4 O que o benchmark não prova

• Não mede PostgreSQL, ORM, Redis, TLS, proxy, container ou rede externa.
• Não mede consumo de memória nem tempo de inicialização.
• Não compara múltiplos processos ou todos os ajustes possíveis.
• Não prova que toda API Robyn será mais rápida que toda API FastAPI.
• Não mede produtividade, ecossistema, debugging ou custo de manutenção.
• Não deve ser comparado diretamente com números de outra máquina.

O benchmark oficial do Robyn e o TechEmpower são referências úteis, mas respondem a configurações próprias. O teste definitivo é uma cópia representativa da sua aplicação, com o mesmo banco, payload, autenticação, observabilidade e padrão de concorrência de produção.

15 FastAPI ou Robyn: como escolher

Escolha Robyn quando o overhead HTTP é mensuravelmente relevante, a equipe aceita um runtime próprio, o escopo pode ser validado em prova de conceito e existe disposição para acompanhar um projeto jovem. Gateways leves, ingestão de eventos, serviços de baixa latência e APIs internas bem delimitadas são bons candidatos.

Escolha FastAPI quando integração ASGI, bibliotecas, exemplos, contratação, ferramentas e previsibilidade operacional pesam mais. APIs com contratos complexos, ecossistema Pydantic consolidado e equipes que já dominam Starlette/Uvicorn normalmente têm menor risco de entrega.

Também existe uma terceira resposta: mantenha Django para o domínio e extraia apenas um serviço crítico quando métricas justificarem. Trocar framework antes de identificar o gargalo costuma mover complexidade sem mover o percentil de latência.

16 Checklist para produção

[ ] Versões de Robyn, Python e dependências estão fixadas?
[ ] Existe wheel para a imagem e arquitetura de produção?
[ ] Handlers async usam apenas I/O realmente assíncrono?
[ ] CPU-bound foi medido e isolado quando necessário?
[ ] Pools de banco consideram todos os processos e réplicas?
[ ] Timeouts existem no cliente, servidor, proxy e dependências?
[ ] Payloads e uploads têm limite?
[ ] Autenticação e autorização foram testadas separadamente?
[ ] CORS lista origens exatas?
[ ] Erros não vazam stack trace ou segredos?
[ ] Logs possuem request ID e dados sensíveis são filtrados?
[ ] Métricas incluem p95, p99, erros e saturação?
[ ] WebSockets/SSE lidam com desconexão e backpressure?
[ ] Estado necessário foi movido para serviço compartilhado?
[ ] Readiness e shutdown gracioso foram testados?
[ ] O benchmark reproduz o tráfego real, não apenas Hello World?
[ ] Existe plano de rollback para upgrades do runtime?

17 Conclusão

Robyn não é apenas uma API Python com marketing de Rust. Seu runtime integrado muda de fato o caminho da requisição e, nos testes deste artigo, reduziu bastante o overhead em endpoints pequenos. A evolução recente também fechou lacunas importantes: Pydantic, OpenAPI, autenticação, TestClient, sub-routers, WebSockets, SSE e streaming tornam o framework capaz de ir além de uma demonstração de velocidade.

Ao mesmo tempo, performance de framework é apenas uma parcela da aplicação. A diferença de 2,53 vezes no JSON mínimo caiu para 1,16 vez quando introduzimos uma espera assíncrona de 10 ms. Em um serviço com banco, cache, rede e domínio, a única forma de saber o ganho é medir o sistema inteiro.

A decisão profissional é equilibrar throughput, latência de cauda, maturidade, ecossistema e capacidade da equipe. Se o runtime é o gargalo, Robyn merece uma prova de conceito séria. Se o risco operacional e a integração pesam mais, FastAPI continua sendo uma escolha excelente. O benchmark abre a conversa; ele não substitui a arquitetura.

18 Referências

• Documentação oficial do Robyn
• Arquitetura oficial do Robyn
• Releases e changelog do Robyn
• Repositório oficial do Robyn no GitHub
• Robyn no PyPI
• TechEmpower Framework Benchmarks
• Documentação oficial do FastAPI
• Documentação oficial do Uvicorn
• Robyn: The Rust-Powered Python Framework That’s Shockingly Fast
• FastAPI vs. Robyn: A Detailed Comparison
• FastAPI vs Robyn, Python API framework comparisons