Skip to main content

Composable mixins for Python services: structured logging with correlation IDs and sensitive data classification.

Project description

mixin-suite

Composable Python mixins for production services: structured logging with automatic correlation-ID propagation, and sensitive-data classification and masking for frozen dataclasses.

PyPI version CI License Python Versions

This distribution consolidates two concern-specific packages and adds composable retry logic:

  • mixin-logging (v0.6.0): End-to-end correlation-ID propagation across 14 adapters for distributed systems
  • mixin-sensitivity (v0.4.0): Decorator-based sensitivity classification and masking for frozen dataclasses
  • mixin-retry: Class-capable retry decorator with full-jitter backoff and predicate-based retry conditions

All packages retain their original import roots (mixin_logging, mixin_sensitivity, mixin_retry) 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 once, and they auto-mask in logs, reprs, and tracebacks.

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 sensitive, classify

@sensitive
@dataclass(frozen=True, slots=True)
class APICredentials:
    user_id: int
    api_token: str = field(metadata={"sensitivity": "secret"})

creds = APICredentials(user_id=1, api_token="sk-abc123xyz")
logger.info("Creds: %s", creds)  # SAFE: repr shows "api_token=***"

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. Decorate and mark sensitive fields

from dataclasses import dataclass, field
from mixin_sensitivity import sensitive, Sensitivity

@sensitive
@dataclass(frozen=True, slots=True)
class User:
    id: int
    api_token: str = field(metadata={"sensitivity": "secret"})
    email: str = field(metadata={"sensitivity": "pii"})
    ssn: str = field(metadata={"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='***', email='***', ssn='***', 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.2.0 distribution.

Logging Example: Correlation-ID Propagation

import logging
from mixin_logging import LoggingMixin, logged, 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."""

    @logged("document.upload")
    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

    @logged("document.process")
    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.2.0):

2026-07-10 03:05:20,405 - __main__.DocumentService - INFO - document.upload.start - correlation_id=req-2026-07-10-001
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 - document.process.start - 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 sensitive, classify, Sensitivity

@sensitive
@dataclass(frozen=True, slots=True)
class HealthRecord:
    patient_id: int
    ssn: str = field(metadata={"sensitivity": "phi"})
    diagnosis: str = field(metadata={"sensitivity": "phi"})
    treatment_notes: str = field(metadata={"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.2.0):

HealthRecord(patient_id=42, ssn=***, diagnosis=***, treatment_notes=***, 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.6.0)

Core classes and functions:

  • LoggingMixin: Base class providing log_info(), log_debug(), and @logged support
  • logged(event_name): Decorator for automatic event logging and error handling
  • 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
  • LoggedClient, LoggedContainer: Internal decorator implementation details
  • PUBLIC_API: Frozenset of all public names

mixin_sensitivity (v0.4.0)

Core classes and functions:

  • sensitive: Class decorator enabling automatic masking of sensitive fields
  • 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

mixin_retry

Core classes and functions:

  • retried(retry_on=None, max_retries=3, base_delay_ms=100): Class-capable decorator for automatic retry logic with exponential backoff
  • RetryClient: Client implementation for retry decorator logic
  • RetryContainer: Container for retry decorator metadata

Imports

All packages maintain their original import roots:

# Logging
from mixin_logging import LoggingMixin, logged, set_correlation_id, get_correlation_id

# Sensitivity
from mixin_sensitivity import sensitive, classify, Sensitivity

# Retry
from mixin_retry import retried

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:

License

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

mixin_suite-0.3.0.tar.gz (392.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

mixin_suite-0.3.0-py3-none-any.whl (82.8 kB view details)

Uploaded Python 3

File details

Details for the file mixin_suite-0.3.0.tar.gz.

File metadata

  • Download URL: mixin_suite-0.3.0.tar.gz
  • Upload date:
  • Size: 392.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for mixin_suite-0.3.0.tar.gz
Algorithm Hash digest
SHA256 3781c5b94eae58e93fd8cc1ca1add142722c09121d149d3a79a6af37a52b8321
MD5 f4137b49e0029f4c444d2c276bbb016b
BLAKE2b-256 82b576b409aeb1e0880df2eb6e8cf6eaf1d80172a9f47299f9ed19f23de24100

See more details on using hashes here.

Provenance

The following attestation bundles were made for mixin_suite-0.3.0.tar.gz:

Publisher: publish.yml on jekhator/mixin-suite

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file mixin_suite-0.3.0-py3-none-any.whl.

File metadata

  • Download URL: mixin_suite-0.3.0-py3-none-any.whl
  • Upload date:
  • Size: 82.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for mixin_suite-0.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 0214060d62e7f59fe24a7b84bebefc80aa6919db9c22ad7d9e3e2f8aa326772d
MD5 18bc5daf7b5a98925674ccfaf08083bc
BLAKE2b-256 a0639454fd39b49b563bf4144acba415e2a2f4ee74e8cce8ca6f46467e27e4cc

See more details on using hashes here.

Provenance

The following attestation bundles were made for mixin_suite-0.3.0-py3-none-any.whl:

Publisher: publish.yml on jekhator/mixin-suite

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page