Integração com Flask
Integre o ImmutableLog em qualquer aplicação Flask com uma extensão que usa os hooks nativos do Flask. Captura automaticamente todas as requisições HTTP — erros, sucessos e exceções — com suporte ao padrão application factory via `init_app`.
Instalação
A extensão depende apenas do Flask e do `requests`. Copie o arquivo `middleware.py` para o diretório da sua aplicação.
pip install flask requestsCódigo da Extensão
Cole o código abaixo em `middleware.py`. A extensão segue o padrão de extensões do Flask, com suporte ao `init_app` para o padrão application factory, e usa `flask.g` para armazenar estado por requisição.
# middleware.py
import hashlib
import json
import logging
import time
import uuid
from datetime import datetime, timezone
import requests as req_lib
from flask import Flask, g, request
logger = logging.getLogger(__name__)
MAX_PAYLOAD_BYTES = 12_000
def clamp_payload(payload: dict) -> str:
s = json.dumps(payload, ensure_ascii=False)
if len(s.encode("utf-8")) <= MAX_PAYLOAD_BYTES:
return s
payload = dict(payload)
payload.pop("error", None)
payload.pop("request_body", None)
s2 = json.dumps(payload, ensure_ascii=False)
if len(s2.encode("utf-8")) <= MAX_PAYLOAD_BYTES:
return s2
return json.dumps({
"id": payload.get("id"),
"kind": payload.get("kind"),
"message": (payload.get("message") or "event")[:500],
"timestamp": payload.get("timestamp"),
}, ensure_ascii=False)
def hash_body(body: bytes) -> str:
return hashlib.sha256(body).hexdigest()
def severity_from(status_code: int | None, exc: Exception | None) -> str:
if exc is not None:
return "error"
if status_code is None:
return "info"
if status_code >= 400:
return "error"
if status_code >= 300:
return "info"
if status_code >= 200:
return "success"
return "info"
def get_severity_level(kind: str, status_code: int | None) -> str:
if kind == "error":
return "high"
elif kind == "warning":
return "medium"
return "low"
class ImmutableLogAudit:
"""Flask extension for ImmutableLog audit logging."""
def __init__(
self,
app: Flask | None = None,
api_key: str = "",
api_url: str = "",
service_name: str = "flask-service",
env: str = "production",
skip_paths: list[str] | None = None,
):
self.api_key = api_key
self.api_url = api_url
self.service_name = service_name
self.env = env
self.skip_paths = set(skip_paths or ["/health", "/healthz"])
if app is not None:
self.init_app(app)
def init_app(self, app: Flask):
app.extensions["immutablelog"] = self
app.before_request(self._before_request)
app.after_request(self._after_request)
app.teardown_request(self._teardown_request)
def _before_request(self):
if request.path in self.skip_paths:
return
g._imtbl_started_at = time.time()
g._imtbl_request_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
g._imtbl_body = request.get_data(cache=True)
g._imtbl_exc_handled = False
def _after_request(self, response):
if request.path in self.skip_paths:
return response
if not getattr(g, "_imtbl_exc_handled", False):
self._emit(response=response, exc=None)
response.headers["X-Request-Id"] = getattr(g, "_imtbl_request_id", "")
return response
def _teardown_request(self, exc):
if exc is not None and not getattr(g, "_imtbl_exc_handled", False):
g._imtbl_exc_handled = True
self._emit(response=None, exc=exc)
def _emit(self, response=None, exc=None):
try:
started = getattr(g, "_imtbl_started_at", None)
latency_ms = int((time.time() - started) * 1000) if started else None
status_code = response.status_code if response else (500 if exc else None)
request_id = getattr(g, "_imtbl_request_id", str(uuid.uuid4()))
body = getattr(g, "_imtbl_body", b"")
kind = severity_from(status_code, exc)
event_name = getattr(g, "imtbl_event_name", None) or f"http.{request.method}.{request.path}"
ip = request.headers.get("X-Forwarded-For", request.remote_addr or "unknown")
ip = ip.split(",")[0].strip()
user_context = {
"ip": ip,
"user_agent": request.user_agent.string or "unknown",
}
payload = {
"id": str(uuid.uuid4()),
"kind": kind,
"message": self._generate_message(status_code, exc, kind),
"timestamp": datetime.now(timezone.utc).isoformat(),
"context": user_context,
"request": {
"request_id": request_id,
"method": request.method,
"path": request.path,
"query_params": dict(request.args) or None,
},
"metrics": {"latency_ms": latency_ms, "status_code": status_code},
"severity": get_severity_level(kind, status_code),
}
if kind == "error":
error_details = {
"status_code": status_code,
"retryable": status_code in [408, 429, 500, 502, 503, 504] if status_code else False,
}
if exc:
error_details["exception"] = type(exc).__name__
error_details["exception_message"] = str(exc)
if body:
error_details["request_body_hash"] = hash_body(body)
if len(body) < 2000:
error_details["request_body"] = body.decode("utf-8", errors="replace")
payload["error"] = error_details
elif kind == "success":
payload["success"] = {"status_code": status_code, "result": "ok" if status_code == 200 else "processed"}
elif kind == "info":
payload["info"] = {"status_code": status_code, "action": request.method.lower()}
meta = {
"type": kind,
"event_name": event_name,
"service": self.service_name,
"request_id": request_id,
"env": self.env,
}
event = {"payload": clamp_payload(payload), "meta": meta}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"Idempotency-Key": f"{event_name}-{request_id}",
"Request-Id": request_id,
"X-Client-TZ": "UTC",
"X-Client-Offset-Minutes": "0",
}
req_lib.post(f"{self.api_url}/v1/events", json=event, headers=headers, timeout=5)
except Exception as e:
logger.warning("ImmutableLogAudit _emit failed: %s", e)
def _generate_message(self, status_code, exc, kind: str) -> str:
method = request.method
path = request.path
if exc:
return f"{method} {path} failed with exception: {type(exc).__name__}"
if kind == "error":
return f"{method} {path} returned error {status_code}"
elif kind == "success":
return f"{method} {path} completed successfully"
return f"{method} {path} processed"Limite: Limite de payload: 12KB por evento. Campos grandes são removidos automaticamente se o limite for ultrapassado.
Configuração e Registro
A extensão suporta dois modos de uso: instanciação direta com o app Flask, ou o padrão application factory com `init_app`. Use o padrão `init_app` em projetos maiores para facilitar testes e modularização.
Modo 1: Instanciação direta
import os
from flask import Flask
from middleware import ImmutableLogAudit
app = Flask(__name__)
ImmutableLogAudit(
app,
api_key=os.environ.get("IMTBL_API_KEY", ""),
api_url=os.environ.get("IMTBL_URL", "https://api.immutablelog.com"),
service_name="meu-servico-flask",
env="production",
)Modo 2: Application Factory (recomendado)
import os
from flask import Flask
from middleware import ImmutableLogAudit
# Instancia sem app — registra depois com init_app
# Instantiate without app — register later with init_app
immutablelog = ImmutableLogAudit(
api_key=os.environ.get("IMTBL_API_KEY", ""),
api_url=os.environ.get("IMTBL_URL", "https://api.immutablelog.com"),
service_name="meu-servico-flask",
env="production",
)
def create_app():
app = Flask(__name__)
immutablelog.init_app(app)
# ... registrar blueprints, etc.
return appNunca hardcode o token. Use variáveis de ambiente e carregue via os.environ.get() ou python-dotenv.
Como funciona
A extensão registra três hooks no Flask. `before_request` captura o timestamp de início, gera o `request_id` e lê o body. `after_request` calcula a latência e emite o evento de sucesso/info/warning. `teardown_request` captura exceções não tratadas e emite o evento de erro com detalhes da exceção.
before_request
Captura timestamp, gera request_id e lê o body via flask.g
after_request
Calcula latência, emite evento e injeta X-Request-Id na resposta
teardown_request
Captura exceções, emite evento de erro e evita double-emit
O flag g._imtbl_exc_handled previne double-emit quando tanto `after_request` quanto `teardown_request` são acionados.
Nomes de evento customizados
Use `g.imtbl_event_name` dentro de qualquer view function ou blueprint para sobrescrever o nome do evento gerado automaticamente. O Flask's `g` garante que o valor é isolado por requisição.
from flask import g, jsonify
@app.post("/checkout")
def process_checkout():
# Sobrescreve o nome do evento para esta view
# Override the event name for this view
g.imtbl_event_name = "payment.checkout.initiated"
# ... logica de negocio / business logic ...
return jsonify({"status": "ok"})Se não definido, o padrão é http.METHOD.path (ex: http.POST./checkout).
Exclusão de rotas
Passe uma lista de paths em `skip_paths` ao instanciar a extensão para ignorar rotas específicas. Ideal para health checks de load balancers e endpoints de métricas.
ImmutableLogAudit(
app,
api_key=os.environ["IMTBL_API_KEY"],
api_url="https://api.immutablelog.com",
# Rotas ignoradas / Ignored routes
skip_paths=["/health", "/healthz", "/metrics", "/ping"],
)Uso com Blueprints
A extensão funciona automaticamente com blueprints, pois os hooks `before_request` e `after_request` são registrados na aplicação principal. Não é necessário nenhuma configuração adicional nos blueprints.
from flask import Blueprint, g, jsonify
payments_bp = Blueprint("payments", __name__, url_prefix="/payments")
@payments_bp.post("/checkout")
def checkout():
# g.imtbl_event_name funciona normalmente em blueprints
# g.imtbl_event_name works normally in blueprints
g.imtbl_event_name = "payment.checkout.initiated"
return jsonify({"status": "ok"})
# app.py
from flask import Flask
from middleware import ImmutableLogAudit
from payments import payments_bp
def create_app():
app = Flask(__name__)
ImmutableLogAudit(app, api_key=..., api_url=...)
app.register_blueprint(payments_bp)
return appEsta documentação reflete o comportamento atual da integração. Para dúvidas ou integrações avançadas, entre em contato com o time de suporte.