Mobius Logging for Python

Python port of the Java Mobius logging library. It standardizes logging context for Mobius FastAPI / Starlette services: a contextvars-based context store (the analog of SLF4J MDC), a request middleware, an @observed method decorator, sensitive-data masking, OpenTelemetry trace sync, JSON logging, and a Kafka schema-event producer.

The Kafka topic, schema id and tenant id come with preconfigured defaults (overridable in setup_logging), so events from Python services land in the same pipeline as the rest of the platform.

Requires Python 3.10+.

Install

pip install mobius-logging-py

# or directly from git
pip install "mobius-logging-py @ git+https://<your-git-host>/mobius-logging-py.git@v1.0.0"

The import package is mobius_logging (e.g. from mobius_logging import setup_logging).

Optional extras: mobius-logging-py[otel] (OpenTelemetry trace sync), mobius-logging-py[fastapi] (Starlette middleware).

Quick start

from fastapi import FastAPI
from mobius_logging import setup_logging, observed, LogType
from py_kafka_producer_client import Producer  # your internal client

app = FastAPI()

setup_logging(
    app,
    service_name="orders-service",
    profile="prod",                       # json logs for staging/prod/production
    sensitive_keys=["session_id", "refresh_token"],
    kafka_producer=Producer(...),         # wrapped in PyKafkaProducerClientSender
)

@app.post("/orders")
@observed(event_type="order.create", log_type=LogType.AUDIT, resource_type="order")
async def create_order(order_id: str):
    ...

setup_logging is the one-call equivalent of Java's Spring auto-configuration (Python has no auto-config, so wiring is explicit). It configures logging, registers custom sensitive keys, builds the schema producer, and adds the request middleware.

Java → Python mapping

Out of scope for v1 (present in Java): JDBC/JPA, MongoDB, and Camunda logging.

Context API

from mobius_logging import context, LogType

context.init()                       # sets schema_version
context.tenant_id("tenant-123")
context.log_type(LogType.AUDIT)
context.resource_type("order")
context.resource_id("order-789")
context.field("custom_key", "value")
context.ensure_correlation_id()      # generates one if missing
context.sync_trace_from_opentelemetry()

snap = context.snapshot()
try:
    context.put("event_type", "order.approved")
finally:
    context.restore(snap)

Note: Java's ensureCorrelationId() has an inverted condition and only regenerates when one already exists. This port fixes that — it generates a correlation id when none is present.

@observed

Works on sync and async functions. Adds log_type, event_type, resource_type, class_name, method_name, outcome, method_duration_ms, and exception_type/exception_message on failure. After completion it ships the context to the schema producer (if one is registered), then restores the pre-call context snapshot (Java clears MDC instead).

@observed(event_type="order.create", resource_type="order")
def create_order(order_id: str): ...

Kafka

Schema event producer

The producer depends only on a small protocol, so the library has no hard Kafka dependency. It matches the py-kafka-producer-client (>=0.1.7) signature — the event is a dict and topic is keyword-only:

class LogEventSender(Protocol):
    def send(self, value: dict, *, topic: str) -> Any: ...

For the internal py-kafka-producer-client, pass the client to setup_logging(kafka_producer=...) or wrap it directly:

from mobius_logging import KafkaLogProducer, PyKafkaProducerClientSender, set_log_producer

producer = KafkaLogProducer(PyKafkaProducerClientSender(client), "orders-service")
set_log_producer(producer)

PyKafkaProducerClientSender calls client.send(value, topic=...), passing the DataIngestionOperation envelope as a dict (the client serializes it). To use a different client, pass any object exposing send(value, *, topic) as kafka_sender=.

Sending runs on a background thread pool (best-effort, never raises, never blocks the request) — the analog of the Java @Async method.

Context propagation

from mobius_logging import producer_headers, apply_consumer_record

# producer side
client.send(topic, value, headers=producer_headers())

# consumer side
apply_consumer_record(record.headers(), topic=record.topic(),
                      partition=record.partition(), offset=record.offset())

Propagated header fields: trace_id, span_id, correlation_id, causation_id, tenant_id. Consumer also sets kafka_topic, kafka_partition, kafka_offset.

Logging output

configure_platform_logging(service_name, profile) installs a root handler that injects the current context onto every record. Profiles staging/prod/ production emit JSON; others use a readable text pattern. A daily-rotating file handler writes to logs/<service>.log (30 days retained).

In JSON mode every context field is included automatically. For the text pattern, txn_id, tenant_id, and trace_id are surfaced inline.

Sensitive data masking

from mobius_logging import mask, mask_map, init_custom_keys

init_custom_keys(["session_id", "refresh_token"])
mask("authorization", token)      # masked
mask_map(context.copy())          # masks all sensitive keys in a dict

Default sensitive keys: password, token, secret, authorization, cookie, apikey, privatekey, kubeconfig. Matching is case-insensitive. Masking keeps the last 4 characters of generic values and partially masks emails — identical behaviour to the Java masker.

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest