Flask Integration
Integrate ImmutableLog into any Flask application with an extension that uses Flask's native hooks. Automatically captures all HTTP requests — errors, successes, and exceptions — with support for the application factory pattern via `init_app`.
Installation
The extension only depends on Flask and `requests`. Copy the `middleware.py` file into your application directory.
pip install flask requestsExtension Code
Paste the code below into `middleware.py`. The extension follows Flask's extension pattern, with `init_app` support for the application factory pattern, and uses `flask.g` to store per-request state.
# 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"Limit: Payload limit: 12KB per event. Large fields are automatically removed if the limit is exceeded.
Configuration and Registration
The extension supports two usage modes: direct instantiation with the Flask app, or the application factory pattern with `init_app`. Use the `init_app` pattern in larger projects to facilitate testing and modularization.
Mode 1: Direct instantiation
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",
)Mode 2: Application Factory (recommended)
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 appNever hardcode the token. Use environment variables and load via os.environ.get() or python-dotenv.
How it works
The extension registers three hooks in Flask. `before_request` captures the start timestamp, generates the `request_id`, and reads the body. `after_request` calculates latency and emits the success/info/warning event. `teardown_request` captures unhandled exceptions and emits the error event with exception details.
before_request
Captures timestamp, generates request_id, and reads body via flask.g
after_request
Calculates latency, emits event, and injects X-Request-Id into response
teardown_request
Captures exceptions, emits error event, and prevents double-emit
The g._imtbl_exc_handled prevents double-emit when both `after_request` and `teardown_request` are triggered.
Custom event names
Use `g.imtbl_event_name` inside any view function or blueprint to override the automatically generated event name. Flask's `g` ensures the value is isolated per request.
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"})If not set, the default is http.METHOD.path (e.g., http.POST./checkout).
Path exclusion
Pass a list of paths in `skip_paths` when instantiating the extension to ignore specific routes. Ideal for load balancer health checks and metrics endpoints.
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"],
)Usage with Blueprints
The extension works automatically with blueprints, as the `before_request` and `after_request` hooks are registered on the main application. No additional configuration is needed in 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 appThis documentation reflects the current integration behavior. For questions or advanced integrations, contact the support team.