Observability Plugin API Reference¶

Structum Observability (structum-observability)¶

Structum Observability provides unified tools for structured logging (JSON/Console), Prometheus metrics, and distributed tracing.

🚀 Installation¶

pip install structum-observability

📚 Full Documentation¶

Documentation, API Reference, and Guides: 👉 https://structum.pages.dev/reference/observability

✨ Key Features¶

• Structured Logging: JSON logs ready for ELK/Loki via structlog.

• Metrics: Native Prometheus metrics exposure.

• Context Awareness: Automatic log enrichment with Context ID and Tracing ID.

Observability Plugin Package¶

Structum Observability Plugin.

Implementation of logging and metrics interfaces using Structlog and contextvars.

structum_lab.plugins.observability.configure_structlog(log_level: str = 'INFO', json_logs: bool = True)[source]¶

Configures the global structlog stack.

class structum_lab.plugins.observability.StructuredLogger(name: str)[source]¶

Bases: object

Wrapper accessible for manual instantiation if needed.

However, the preferred way is using get_logger(). This class is merely a compatibility shim or provider access point.

__init__(name: str)[source]¶

Initialize the structured logger.

Parameters:¶

name: str¶

Logger name.

bind(**kwargs: Any) → StructuredLogger[source]¶

Bind context variables to a new logger instance.

critical(message: str, **kwargs: Any) → None[source]¶

Log a critical message.

debug(message: str, **kwargs: Any) → None[source]¶

Log a debug message.

error(message: str, **kwargs: Any) → None[source]¶

Log an error message.

info(message: str, **kwargs: Any) → None[source]¶

Log an info message.

warning(message: str, **kwargs: Any) → None[source]¶

Log a warning message.

class structum_lab.plugins.observability.MetricsCollector[source]¶

Bases: object

Raccolta metriche in-memory (simil-Prometheus/StatsD).

__init__()[source]¶

Initialize the metrics collector with empty storage.

get_counter(metric: str, labels: dict[str, str] | None = None) → int[source]¶

Get the current value of a counter (for introspection).

increment(metric: str, labels: dict[str, str] | None = None) → None[source]¶

Increment a counter metric.

observe(metric: str, value: float, labels: dict[str, str] | None = None) → None[source]¶

Record an observation value (e.g., duration).

structum_lab.plugins.observability.track_operation(operation_name: str)[source]¶

Decorator per tracciare durata e successo/errore di un’operazione.

structum_lab.plugins.observability.initialize(log_level: str = 'INFO', json_logs: bool = True)[source]¶

Manual initialization function.

Metrics¶

In-Memory Metrics Collection.

Provides a simple in-memory metrics collector for development and testing, along with a track_operation decorator for automatic timing.

class structum_lab.plugins.observability.metrics.MetricsCollector[source]¶

Bases: object

Raccolta metriche in-memory (simil-Prometheus/StatsD).

__init__()[source]¶

Initialize the metrics collector with empty storage.

increment(metric: str, labels: dict[str, str] | None = None) → None[source]¶

Increment a counter metric.

observe(metric: str, value: float, labels: dict[str, str] | None = None) → None[source]¶

Record an observation value (e.g., duration).

get_counter(metric: str, labels: dict[str, str] | None = None) → int[source]¶

Get the current value of a counter (for introspection).

structum_lab.plugins.observability.metrics.track_operation(operation_name: str)[source]¶

Decorator per tracciare durata e successo/errore di un’operazione.

Observability Core¶

Structlog Integration for Structum.

This module configures the structlog library to provide enterprise-grade structured logging, compliant with the LoggerInterface protocol.

structum_lab.plugins.observability.observability.configure_structlog(log_level: str = 'INFO', json_logs: bool = True)[source]¶

Configures the global structlog stack.

class structum_lab.plugins.observability.observability.StructuredLogger(name: str)[source]¶

Bases: object

Wrapper accessible for manual instantiation if needed.

However, the preferred way is using get_logger(). This class is merely a compatibility shim or provider access point.

__init__(name: str)[source]¶

Initialize the structured logger.

Parameters:¶

name: str¶

Logger name.

debug(message: str, **kwargs: Any) → None[source]¶

Log a debug message.

info(message: str, **kwargs: Any) → None[source]¶

Log an info message.

warning(message: str, **kwargs: Any) → None[source]¶

Log a warning message.

error(message: str, **kwargs: Any) → None[source]¶

Log an error message.

critical(message: str, **kwargs: Any) → None[source]¶

Log a critical message.

bind(**kwargs: Any) → StructuredLogger[source]¶

Bind context variables to a new logger instance.

structum_lab.plugins.observability.observability.get_structlog_logger(name: str) → LoggerInterface[source]¶

Factory function to replace structum_lab.logging.get_logger.

Prometheus Backend¶

Prometheus Metrics Implementation.

Provides a concrete implementation of MetricsInterface using prometheus_client.

class structum_lab.plugins.observability.prometheus.PrometheusMetrics(namespace: str = 'structum')[source]¶

Bases: MetricsInterface

Prometheus implementation of MetricsInterface.

__init__(namespace: str = 'structum')[source]¶

Initialize the Prometheus metrics exporter.

Parameters:¶

namespace: str = 'structum'¶

Prefix for all metric names.

Raises:¶

ImportError – If prometheus_client is not installed.

increment(name: str, value: float = 1.0, tags: dict[str, str] | None = None) → None[source]¶

Increment a counter.

gauge(name: str, value: float, tags: dict[str, str] | None = None) → None[source]¶

Set a gauge value.

timing(name: str, value: float, tags: dict[str, str] | None = None) → None[source]¶

Record a timing (uses histogram).

histogram(name: str, value: float, tags: dict[str, str] | None = None) → None[source]¶

Record a value in histogram.