mixin-suite

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.

What They Do

Logging: Correlation-ID Propagation

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 context

After:

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 automatically

Sensitivity: Prevent Accidental Secret Leaks

Mark 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 exposed

After:

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***"

Retry: Exponential Backoff with Predicates

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")

Latency: High-Precision Measurement

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 exit

Installation

Base installation:

pip install mixin-suite

or with uv:

uv add mixin-suite

With 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 adapters

Python version: Requires Python 3.11 or later (3.11 and 3.14 tested).

Quick Start

Logging with Correlation IDs

1. Add stdlib adapter to your logging config

import logging
from mixin_logging.adapters.stdlib.stdlib_client import CorrelationLogFilter

logging.basicConfig()
logging.getLogger().addFilter(CorrelationLogFilter())

2. Install FastAPI adapter and add middleware

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)

3. Use LoggingMixin in your classes

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 automatically

Sensitivity: Masking Sensitive Fields

1. Mark sensitive fields and inherit SensitiveRepr

from 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: str

2. Use in your code

user = 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),
# ))

Run-Verified Examples

These examples are executed against the published mixin-suite==0.5.0 distribution.

Logging Example: Correlation-ID Propagation

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

Sensitivity Example: Data Masking

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']

Documentation

• 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.md and docs/mixin_sensitivity/CHANGELOG-history.md

Public API

mixin_logging (v0.5.0)

Core classes and functions:

• LoggingMixin: Base class providing log_info(), log_debug(), log_warning(), log_error(), and log_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

mixin_sensitivity (v0.5.0)

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

mixin_retry (v0.5.0)

Core classes and functions:

• RetryPolicy: Configuration object for retry behavior (max_attempts, backoff, jitter, predicates)
• RetryExecutor: Client for wrapping functions with retry logic via the wrap(operation, /, policy) method

Imports

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, SuppressionPolicy

Contributing

This 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

License

Licensed under the Apache License 2.0. See LICENSE for details.