Composable Python mixins for production services: structured logging with automatic correlation-ID propagation, sensitive-data masking, retry logic, and latency measurement.
This distribution includes five composable roots:
- mixin_logging: End-to-end correlation-ID propagation, LoggingMixin, ambient logging, FlushOnWarningHandler
- mixin_sensitivity: Sensitivity classification and dataclass repr-masking via SensitiveRepr
- mixin_retry: Exponential backoff retry logic via RetryPolicy/RetryExecutor (capability contracts)
- mixin_latency: High-precision elapsed-time measurement via LatencyClock
- mixin_notifications: Event dispatch and suppression across multi-step workflows
All packages retain their original import roots and can be used independently or together.
Track a single request through a distributed system with automatic correlation-ID injection on every log, HTTP call, database query, and background task.
Before:
class OrderService:
def create_order(self, order_id: int):
print(f"Creating order {order_id}") # No correlation tracking
send_notification(order_id) # Loses request contextAfter:
from mixin_logging import LoggingMixin, set_correlation_id
set_correlation_id("req-123")
class OrderService(LoggingMixin):
def create_order(self, order_id: int):
self.log_info("order.create", order_id=order_id)
# Logs with: {"correlation_id": "req-123", "order_id": 123, ...}
send_notification(order_id) # Correlation ID propagates automaticallyMark sensitive fields in dataclasses via field metadata, and adopt SensitiveRepr to auto-mask in repr output.
Before:
from dataclasses import dataclass
@dataclass(frozen=True)
class APICredentials:
user_id: int
api_token: str
creds = APICredentials(user_id=1, api_token="sk-abc123xyz")
logger.info("Creds: %s", creds) # LEAKED: api_token exposedAfter:
from dataclasses import dataclass, field
from mixin_sensitivity import Sensitivity, SensitiveRepr
@dataclass(frozen=True, slots=True, repr=False)
class APICredentials(SensitiveRepr):
user_id: int
api_token: str = field(metadata={"sensitivity": Sensitivity.SECRET})
creds = APICredentials(user_id=1, api_token="sk-abc123xyz")
logger.info("Creds: %s", creds) # SAFE: repr shows "api_token=***MASKED***"Resilient function execution with configurable backoff and predicate-based retry decisions.
from mixin_retry import RetryPolicy, RetryExecutor
policy = RetryPolicy(
max_attempts=3,
backoff_base_seconds=0.1,
backoff_multiplier=2.0,
backoff_max_seconds=1.0,
jitter=True,
should_retry=lambda exc: isinstance(exc, ConnectionError)
)
executor = RetryExecutor()
def flaky_api_call(url):
# Retries on ConnectionError, exponential backoff
pass
wrapped_call = executor.wrap(flaky_api_call, policy=policy)
result = wrapped_call("https://api.example.com")Measure elapsed time with perf_counter precision and automatic rounding.
from mixin_latency import LatencyClock
clock = LatencyClock.start()
# ... do work ...
measurement = clock.stop()
print(f"Elapsed: {measurement.duration_ms} ms")
# Or context-manager form:
with LatencyClock.measure() as clock:
# ... do work ...
pass # Duration auto-measured on exitBase installation:
pip install mixin-suiteor with uv:
uv add mixin-suiteWith optional extras for logging adapters:
Base package includes mixin_logging (stdlib adapter only) and mixin_sensitivity (no dependencies).
Optional extras (mixin_logging adapters):
-
[aiohttp]: aiohttp client instrumentation -
[botocore]: AWS SDK instrumentation -
[celery]: Celery task propagation -
[fastapi]: FastAPI middleware and dependencies -
[grpc]: gRPC server instrumentation -
[httpx]: HTTPX client instrumentation -
[requests]: Requests client instrumentation -
[urllib3]: urllib3 client instrumentation -
[all]: All adapters
Install with extras:
uv add "mixin-suite[httpx,botocore]" # Multiple extras
uv add "mixin-suite[all]" # All adaptersPython version: Requires Python 3.11 or later (3.11 and 3.14 tested).
import logging
from mixin_logging.adapters.stdlib.stdlib_client import CorrelationLogFilter
logging.basicConfig()
logging.getLogger().addFilter(CorrelationLogFilter())For FastAPI applications:
from fastapi import FastAPI
from mixin_logging.adapters.fastapi import CorrelationIdMiddleware
app = FastAPI()
app.add_middleware(CorrelationIdMiddleware)Or manually for other frameworks:
from mixin_logging import set_correlation_id
from fastapi import Request
@app.middleware("http")
async def correlation_middleware(request: Request, call_next):
set_correlation_id(request.headers.get("x-correlation-id", "auto-gen"))
return await call_next(request)from mixin_logging import LoggingMixin
class UserService(LoggingMixin):
def create_user(self, user_name: str):
self.log_info("user.create", user_name=user_name)
# Logs include correlation_id automaticallyfrom dataclasses import dataclass, field
from mixin_sensitivity import SensitiveRepr, Sensitivity
@dataclass(frozen=True, slots=True, repr=False)
class User(SensitiveRepr):
id: int
api_token: str = field(metadata={"sensitivity": Sensitivity.SECRET})
email: str = field(metadata={"sensitivity": Sensitivity.PII})
ssn: str = field(metadata={"sensitivity": Sensitivity.PHI})
name: struser = User(
id=1,
api_token="sk-123456",
email="alice@example.com",
ssn="123-45-6789",
name="Alice"
)
# Safe for logging
logger.info("User created: %s", repr(user))
# → "User created: User(id=1, api_token=***MASKED***, email=***MASKED***, ssn=***MASKED***, name='Alice')"
# Introspect sensitivity profile
from mixin_sensitivity import classify
profile = classify(user)
# → SensitivityProfile(classes=(
# ('api_token', Sensitivity.SECRET),
# ('email', Sensitivity.PII),
# ('ssn', Sensitivity.PHI),
# ))These examples are executed against the published mixin-suite==0.5.0 distribution.
import logging
from mixin_logging import LoggingMixin, set_correlation_id
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s - correlation_id=%(correlation_id)s",
)
class DocumentService(LoggingMixin):
"""Service that processes documents with correlation-ID tracking."""
def upload(self, doc_name: str, size_bytes: int) -> dict:
"""Upload a document and return metadata."""
self.log_info("upload.initiated", doc_name=doc_name, size_bytes=size_bytes)
result = {"id": "doc-123", "doc_name": doc_name, "stored": True}
self.log_info("upload.complete", doc_id=result["id"])
return result
def process(self, doc_id: str) -> str:
"""Process a document and return status."""
self.log_info("process.started", doc_id=doc_id)
status = "processed"
self.log_info("process.finished", doc_id=doc_id, status=status)
return status
# Execute with correlation context
set_correlation_id("req-2026-07-10-001")
service = DocumentService()
result = service.upload("report.pdf", 1024000)
status = service.process("doc-123")Output (Python 3.14, mixin-suite==0.5.0):
2026-07-10 03:05:20,405 - __main__.DocumentService - INFO - upload.initiated - correlation_id=req-2026-07-10-001
2026-07-10 03:05:20,405 - __main__.DocumentService - INFO - upload.complete - correlation_id=req-2026-07-10-001
2026-07-10 03:05:20,405 - __main__.DocumentService - INFO - process.started - correlation_id=req-2026-07-10-001
2026-07-10 03:05:20,405 - __main__.DocumentService - INFO - process.finished - correlation_id=req-2026-07-10-001
from dataclasses import dataclass, field
from mixin_sensitivity import SensitiveRepr, classify, Sensitivity
@dataclass(frozen=True, slots=True, repr=False)
class HealthRecord(SensitiveRepr):
patient_id: int
ssn: str = field(metadata={"sensitivity": Sensitivity.PHI})
diagnosis: str = field(metadata={"sensitivity": Sensitivity.PHI})
treatment_notes: str = field(metadata={"sensitivity": Sensitivity.PHI})
attending_physician: str
record = HealthRecord(
patient_id=42,
ssn="987-65-4321",
diagnosis="Type 2 Diabetes",
treatment_notes="Prescribed Metformin 500mg",
attending_physician="Dr. Smith"
)
# Safe repr
print(repr(record))
# Introspect profile
profile = classify(record)
print(f"PHI fields: {[f for f, s in profile.classes if s == Sensitivity.PHI]}")Output (Python 3.14, mixin-suite==0.5.0):
HealthRecord(patient_id=42, ssn=***MASKED***, diagnosis=***MASKED***, treatment_notes=***MASKED***, attending_physician='Dr. Smith')
PHI fields: ['ssn', 'diagnosis', 'treatment_notes']
-
Logging: See
docs/mixin_logging/for detailed adapter documentation, architecture, and integration patterns -
Sensitivity: See
docs/mixin_sensitivity/for classifier API, masking customization, and examples -
Historical Changelogs: See
docs/mixin_logging/CHANGELOG-history.mdanddocs/mixin_sensitivity/CHANGELOG-history.md
Core classes and functions:
-
LoggingMixin: Base class providinglog_info(),log_debug(),log_warning(),log_error(), andlog_exception()methods -
set_correlation_id(id): Set the correlation ID for the current context -
get_correlation_id(): Retrieve the current correlation ID -
clear_correlation_id(): Clear the correlation ID from context -
CorrelationContext: Data class representing correlation metadata -
ContextVarClient: Internal context-variable manager for correlation propagation -
FlushOnWarningHandler: Logging handler that flushes on WARNING level or above -
FlushOnWarningConfig: Configuration for the flush-on-warning handler -
AmbientLogger: Namespace for ambient logging functions (log_info, log_debug, log_warning, log_error) -
PUBLIC_API: Frozenset of all public names
Core classes and functions:
-
SensitiveRepr: Base class for dataclasses that masks sensitive fields in repr output -
classify(dataclass_or_instance): Introspect sensitivity profile of a dataclass -
Sensitivity: Enum taxonomy: PHI, PII, PCI, SECRET -
SensitivityProfile: Data class containing field-to-sensitivity mappings -
SensitiveDeclarationError: Exception raised for invalid sensitivity declarations
Core classes and functions:
-
RetryPolicy: Configuration object for retry behavior (max_attempts, backoff, jitter, predicates) -
RetryExecutor: Client for wrapping functions with retry logic via thewrap(operation, /, policy)method
All packages maintain their original import roots:
# Logging
from mixin_logging import LoggingMixin, set_correlation_id, get_correlation_id
# Sensitivity
from mixin_sensitivity import SensitiveRepr, classify, Sensitivity
# Retry
from mixin_retry import RetryPolicy, RetryExecutor
# Latency
from mixin_latency import LatencyClock
# Notifications
from mixin_notifications import Dispatcher, SuppressionPolicyThis is a consolidation of independently-maintained packages. Bug reports and feature requests should be filed in this repository or the respective upstream repositories:
- mixin-logging historical issues: https://github.com/jekhator/mixin-logging/issues
- mixin-sensitivity historical issues: https://github.com/jekhator/mixin-sensitivity/issues
- mixin-retry and suite issues: https://github.com/jekhator/mixin-suite/issues
Licensed under the Apache License 2.0. See LICENSE for details.
