mobius-logging-py 1.0.3

Standardized logging context, observation, masking and Kafka schema events for Mobius Python (FastAPI) services.

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

Java Mobius logging library	Python (mobius_logging)
SLF4J MDC	contextvars.ContextVar (context.py)
PlatformLogContext	mobius_logging.context functions
PlatformLogFields	mobius_logging.fields
LogType enum	LogType enum
PlatformLoggingFilter (servlet)	PlatformLoggingMiddleware (ASGI)
@PlatformObserved + AOP aspect	@observed decorator
KafkaLogProducer (KafkaTemplate)	KafkaLogProducer + LogEventSender
PlatformKafka*Context	producer_headers / apply_consumer_record
SensitiveDataMasker	mobius_logging.masking
LogbackConfigurator	configure_platform_logging
Spring auto-config classes	setup_logging (explicit)

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