Extensão Flask que captura automaticamente todas as requisições HTTP — erros, sucessos e exceções não tratadas — e envia eventos de auditoria imutáveis para o ImmutableLog.
Segue o padrão de extensões Flask com suporte nativo ao init_app para o padrão application factory.
A única dependência adicional é o requests (Flask já está no projeto):
pip install requestsOpção A — Copiar o arquivo:
Copie middleware.py para dentro do seu projeto:
app/extensions.py
core/middleware.py
ext/immutablelog.py
Opção B — Instalar em modo editável:
pip install -e /caminho/para/integrations_immutablelog/immutablelog_flask_middleware| Variável | Obrigatório | Default | Descrição |
|---|---|---|---|
IMTBL_API_KEY |
Sim | — | Chave de API do ImmutableLog |
IMTBL_URL |
Sim | — | URL base da API |
IMTBL_SERVICE_NAME |
Não | flask-service |
Nome do serviço exibido nos eventos |
IMTBL_ENV |
Não | production |
Ambiente: production, staging, development |
Segurança: nunca exponha
IMTBL_API_KEYem código ou repositórios. Use sempre variáveis de ambiente.
Modo 1 — Instanciação direta:
# app.py
import os
from flask import Flask
from app.extensions import ImmutableLogAudit # Opção A
# from immutablelog_flask_middleware.middleware import ImmutableLogAudit # Opção B
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=os.environ.get("IMTBL_SERVICE_NAME", "flask-service"),
env=os.environ.get("IMTBL_ENV", "production"),
skip_paths=["/health", "/healthz"], # opcional
)Modo 2 — Application Factory com init_app (recomendado):
# extensions.py
import os
from middleware import ImmutableLogAudit
immutablelog = ImmutableLogAudit(
api_key=os.environ.get("IMTBL_API_KEY", ""),
api_url=os.environ.get("IMTBL_URL", "https://api.immutablelog.com"),
service_name=os.environ.get("IMTBL_SERVICE_NAME", "flask-service"),
env=os.environ.get("IMTBL_ENV", "production"),
)
# app/__init__.py
from flask import Flask
from .extensions import immutablelog
def create_app():
app = Flask(__name__)
immutablelog.init_app(app)
# ... registrar blueprints, etc.
return appCom python-dotenv:
# app.py (início do arquivo)
from dotenv import load_dotenv
from pathlib import Path
load_dotenv(Path(__file__).resolve().parent / ".env")Exemplo de .env:
IMTBL_API_KEY=iml_live_xxxxxxxxxxxxxxxx
IMTBL_URL=https://api.immutablelog.com
IMTBL_SERVICE_NAME=meu-servico-flask
IMTBL_ENV=production| Parâmetro | Tipo | Obrigatório | Default | Descrição |
|---|---|---|---|---|
app |
Flask | None |
Não | None |
Instância do Flask (use init_app se None) |
api_key |
str |
Sim | — | Chave de API do ImmutableLog |
api_url |
str |
Sim | — | URL base da API |
service_name |
str |
Não | flask-service |
Nome do serviço nos eventos |
env |
str |
Não | production |
Ambiente de execução |
skip_paths |
list[str] |
Não | ["/health", "/healthz"] |
Paths que não geram eventos |
A extensão registra três hooks no Flask que cobrem o ciclo de vida completo de cada requisição:
| Hook | Responsabilidade |
|---|---|
before_request |
Captura timestamp de início, gera request_id e lê o body via flask.g |
after_request |
Calcula latência, emite evento de sucesso/info/warning e injeta X-Request-Id |
teardown_request |
Captura exceções não tratadas, emite evento de erro e previne double-emit |
O flag g._imtbl_exc_handled garante que apenas um evento seja emitido por requisição, mesmo quando after_request e teardown_request são acionados em sequência.
Por padrão o evento é nomeado http.{METHOD}.{path} (ex: http.POST./checkout). Para sobrescrever em uma view ou blueprint:
from flask import g, jsonify
@app.post("/checkout")
def process_checkout():
g.imtbl_event_name = "payment.checkout.initiated"
# ... lógica de negócio ...
return jsonify({"status": "ok"})O flask.g garante isolamento por requisição — cada request tem seu próprio contexto.
Passe uma lista de paths em skip_paths ao instanciar a extensão:
ImmutableLogAudit(
app,
api_key=os.environ["IMTBL_API_KEY"],
api_url="https://api.immutablelog.com",
skip_paths=["/health", "/healthz", "/metrics", "/ping"],
)A extensão funciona automaticamente com blueprints sem nenhuma configuração adicional, pois os hooks são registrados na aplicação principal:
from flask import Blueprint, g, jsonify
payments_bp = Blueprint("payments", __name__, url_prefix="/payments")
@payments_bp.post("/checkout")
def checkout():
g.imtbl_event_name = "payment.checkout.initiated"
return jsonify({"status": "ok"})
# app/__init__.py
def create_app():
app = Flask(__name__)
ImmutableLogAudit(app, api_key=..., api_url=...)
app.register_blueprint(payments_bp)
return appCada evento enviado ao ImmutableLog segue esta estrutura:
{
"payload": "{\"id\":\"uuid\",\"kind\":\"success\",\"message\":\"POST /checkout completed successfully\",\"timestamp\":\"2026-02-21T12:00:00Z\",\"context\":{\"ip\":\"1.2.3.4\",\"user_agent\":\"...\"},\"request\":{\"request_id\":\"uuid\",\"method\":\"POST\",\"path\":\"/checkout\",\"query_params\":null},\"metrics\":{\"latency_ms\":31,\"status_code\":200},\"severity\":\"low\"}",
"meta": {
"type": "success",
"event_name": "payment.checkout.initiated",
"service": "flask-service",
"request_id": "uuid",
"env": "production"
}
}O campo
payloadé uma string JSON serializada — não um objeto. Isso garante que o hash SHA-256 seja calculado sobre exatamente o que foi enviado.
| Situação | Resultado |
|---|---|
api_key ou api_url vazios |
Evento não enviado (falha silenciosa, sem exceção) |
Path em skip_paths |
Evento ignorado, requisição prossegue normalmente |
| Status 2xx | kind: success, severity: low |
| Status 3xx | kind: info, severity: low |
| Status 4xx / 5xx | kind: error, severity: high |
| Exceção não tratada | kind: error com exception e exception_message |
| Payload > 12KB | Campos error e request_body removidos automaticamente |
| Erro no envio ao ImmutableLog | Log de warning; requisição não é afetada |