API Reference

Protocols

AlertDispatcherProtocol

source

Protocol for dispatching operational alerts.

Implementations route alerts to notification channels (logging, PagerDuty, Slack, etc.). lexigram-monitor ships a built-in LoggingAlertDispatcher that writes alerts to the structured logger.

Example

class SlackAlertDispatcher:
    async def send_alert(
        self,
        title: str,
        message: str,
        severity: str,
        context: dict[str, Any] | None = None,
    ) -> None:
        await self._slack.post(
            channel="#ops",
            text=f"[{severity}] {title}: {message}",
        )

class SlackAlertDispatcher:
    async def send_alert(
        self,
        title: str,
        message: str,
        severity: str,
        context: dict[str, Any] | None = None,
    ) -> None:
        await self._slack.post(
            channel="#ops",
            text=f"[{severity}] {title}: {message}",
        )

send_alert

async def send_alert(
    title: str,
    message: str,
    severity: str,
    context: dict[str, Any] | None = None
) -> None

source

Dispatch a free-form operational alert.

Parameters

Parameter	Type	Description
`title`	str	Short, human-readable alert title.
`message`	str	Detailed alert message.
`severity`	str	Severity level string, e.g. ``"low"``, ``"high"``, ``"critical"``.
`context`	dict[str, Any] | None	Optional free-form mapping of additional metadata.

send_metric_alert

async def send_metric_alert(
    metric_name: str,
    current_value: float,
    threshold: float,
    context: dict[str, Any] | None = None
) -> None

source

Dispatch an alert triggered by a metric threshold breach.

Parameters

Parameter	Type	Description
`metric_name`	str	Name of the metric that breached its threshold.
`current_value`	float	Observed metric value at the time of the alert.
`threshold`	float	The configured threshold that was exceeded.
`context`	dict[str, Any] | None	Optional free-form mapping of additional metadata.

---

HealthCheckerProtocol

source

Structural contract for a single health-check component.

Any object implementing check() satisfies this protocol, regardless of inheritance from any concrete class.

check

async def check() -> HealthCheckResult

source

Perform the health check and return a result.

---

MetricProtocol

source

Protocol for metric implementations.

Metrics track numeric measurements over time.

name

property name() -> str

source

MetricProtocol name.

description

property description() -> str

source

MetricProtocol description.

record

def record(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

Record a metric value.

Parameters

Parameter	Type	Description
`value`	float	Numeric value to record.
`labels`	dict[str, str] | None	Optional labels/tags.

---

MetricsBackendProtocol

source

Protocol for metrics backend implementations (metrics only).

Backends export metrics to external systems.

initialize

async def initialize() -> None

source

Initialize the metrics backend.

shutdown

async def shutdown() -> None

source

Shutdown the metrics backend.

record_metric

def record_metric(
    name: str,
    value: Any,
    metric_type: str,
    labels: dict[str, str] | None = None
) -> None

source

Record a metric value.

Parameters

Parameter	Type	Description
`name`	str	MetricProtocol name.
`value`	Any	MetricProtocol value.
`metric_type`	str	Type of metric (counter, gauge, histogram).
`labels`	dict[str, str] | None	Optional labels.

---

SpanExporter

source

export

def export(spans: list[Span]) -> None

source

---

Classes

AlertFiredHook

source

Payload fired when a monitoring alert is triggered.

Attributes: alert_name: Name of the alert rule that fired. severity: Severity level (e.g. "warning", "critical").

---

BackendType

source

Supported monitoring backend types.

---

BufferedMetricEntry

source

A single buffered metric recording.

Attributes: name: MetricProtocol name. value: Numeric value to record. labels: Optional label key-value pairs. timestamp: Monotonic timestamp when the entry was created.

---

BufferedMetricRecorder

source

Buffers metric recordings and flushes them in configurable batches.

Reduces metric backend overhead for high-throughput code paths by accumulating entries and flushing at intervals or when the buffer is full.

Parameters

Parameter	Type	Description
`backend`		MetricProtocol backend instance with metric attributes exposing ``record()``.
`flush_interval`		Seconds between automatic periodic flushes. Default: 5.0.
`max_buffer_size`		Maximum entries before a forced flush. Default: 1000.

Example

recorder = BufferedMetricRecorder(backend=monitor_backend, flush_interval=2.0)
await recorder.start()

recorder.record("request_count", 1.0, {"method": "GET"})
# ... many more records ...

await recorder.stop()  # flushes remaining entries

recorder = BufferedMetricRecorder(backend=monitor_backend, flush_interval=2.0)
await recorder.start()

recorder.record("request_count", 1.0, {"method": "GET"})
# ... many more records ...

await recorder.stop()  # flushes remaining entries

__init__

def __init__(
    backend: Any,
    flush_interval: float = 5.0,
    max_buffer_size: int = 1000
) -> None

source

record

def record(
    name: str,
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

Buffer a metric entry for deferred recording.

If the buffer reaches max_buffer_size, schedules an immediate flush.

Parameters

Parameter	Type	Description
`name`	str	MetricProtocol name (must match an attribute on the backend).
`value`	float	Numeric value to record.
`labels`	dict[str, str] | None	Optional label key-value pairs.

flush

async def flush() -> int

source

Flush all buffered entries to the backend.

Returns

Type	Description
int	Number of entries successfully flushed.

start

async def start() -> None

source

Start the periodic background flush task.

stop

async def stop() -> None

source

Stop the background flush task and flush remaining entries.

---

ConsoleMetricsExporterHandler

source

Handler for console metrics exporter.

Creates a ConsoleMetricExporter for outputting metrics to the console.

can_handle

def can_handle(exporter_type: str) -> bool

source

Check if this handler can handle the exporter type.

Parameters

Parameter	Type	Description
`exporter_type`	str	The type of exporter to check.

Returns

Type	Description
bool	True if exporter_type is "console", False otherwise.

create_exporter

def create_exporter(exp_config: Any) -> Any

source

Create a console metrics exporter instance.

Parameters

Parameter	Type	Description
`exp_config`	Any	Configuration for the exporter (unused for console).

Returns

Type	Description
Any	A ConsoleMetricExporter instance.

---

ConsoleSpanExporter

source

export

def export(spans: list[Span]) -> None

source

---

ConsoleTracingExporterHandler

source

Handler for console tracing exporter.

Creates a ConsoleSpanExporter for outputting traces to the console.

can_handle

def can_handle(exporter_type: str) -> bool

source

Check if this handler can handle the exporter type.

Parameters

Parameter	Type	Description
`exporter_type`	str	The type of exporter to check.

Returns

Type	Description
bool	True if exporter_type is "console", False otherwise.

create_exporter

def create_exporter(exp_config: Any) -> Any

source

Create a console tracing exporter instance.

Parameters

Parameter	Type	Description
`exp_config`	Any	Configuration for the exporter (unused for console).

Returns

Type	Description
Any	A ConsoleSpanExporter instance.

---

Counter

source

Monotonically increasing counter.

__init__

def __init__(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
)

source

name

property name() -> str

source

description

property description() -> str

source

record

def record(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

increment

def increment(
    amount: int = 1,
    labels: dict[str, str] | None = None
) -> None

source

Alias for record — delegates to the canonical implementation.

get_count

def get_count() -> int

source

reset

def reset() -> None

source

Reset the counter to zero and clear the observation history.

---

FunctionHealthCheck

source

Health check implemented as a function.

__init__

def __init__(
    name: str,
    func: Callable,
    critical: bool = True
)

source

check

async def check() -> HealthCheckResult

source

---

FunctionProfileResult

source

Result of profiling a function.

---

Gauge

source

Gauge that can go up and down.

__init__

def __init__(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
)

source

name

property name() -> str

source

description

property description() -> str

source

record

def record(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

set

def set(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

Alias for record — delegates to the canonical implementation.

increment

def increment(
    amount: float = 1,
    labels: dict[str, str] | None = None
) -> None

source

Alias for record — increments gauge by amount via canonical implementation.

decrement

def decrement(
    amount: float = 1,
    labels: dict[str, str] | None = None
) -> None

source

Alias for record — decrements gauge by amount via canonical implementation.

get_value

def get_value() -> float

source

reset

def reset() -> None

source

Reset the gauge to zero and clear the observation history.

---

HealthCheck

source

Base class for health checks.

__init__

def __init__(
    name: str,
    critical: bool = True
)

source

check

async def check() -> HealthCheckResult

source

---

HealthCheckCategory

source

Categorises health checks by their role in the Kubernetes health model.

Attributes: LIVENESS: Checks that the application is alive (not deadlocked). Used by /health/live endpoints. Failure triggers a restart. READINESS: Checks that the application is ready to accept traffic. Used by /health/ready endpoints. Failure removes the instance from the load-balancer rotation. STARTUP: Checks that the application has finished its startup sequence. Used by /health/startup endpoints. Succeeded once; after that the probe switches to liveness/readiness.

---

HealthCheckConfig

source

Configuration for health checks.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

Attributes: enabled: Whether health checks are enabled. path: HTTP path for health endpoint. include_details: Include detailed component health in response. timeout: Timeout for health check operations (seconds). checks: List of health check names to include.

---

HealthCheckProvider

source

ASGI middleware that exposes a ``GET /health`` liveness/readiness endpoint.

Intercepts requests to path and returns a JSON health-status payload. All other requests are passed through with a 404 response (or forwarded to the next ASGI app when composed in a middleware stack).

The HTTP status code reflects the overall application health:

• 200 — all checks healthy
• 207 — at least one check degraded
• 503 — at least one check unhealthy

To add custom checks, subclass HealthCheckProvider and override check_health.

Parameters

Parameter	Type	Description
`path`		URL path for the health endpoint. Defaults to ``"/health"``.

Example

from lexigram.monitor.middleware import HealthCheckProvider

app = HealthCheckProvider(path="/ready")
# Mount *app* in your ASGI server / router.

from lexigram.monitor.middleware import HealthCheckProvider

app = HealthCheckProvider(path="/ready")
# Mount *app* in your ASGI server / router.

__init__

def __init__(path: str = '/health') -> None

source

check_health

async def check_health() -> dict

source

Return the current application health as a serialisable mapping.

Override this method in a subclass to perform real dependency checks (database ping, cache ping, etc.).

Returns

Type	Description
dict	A ``dict`` with at minimum a ``"status"`` key (one of ``"healthy"``, ``"degraded"``, or ``"unhealthy"``) and a ``"timestamp"`` float. Additional ``"checks"`` entries can be added by subclasses.

---

HealthCheckRegistry

source

Registry for managing health checks.

Extends Registry for unified introspection. Sidecar lists track which checks participate in liveness vs readiness probes.

__init__

def __init__() -> None

source

add

def add(
    name: str,
    check: Callable[[], Any],
    *,
    timeout: float | None = None,
    critical: bool = True,
    category: Any = None
) -> None

source

Add a health check to satisfy HealthCheckRegistryProtocol.

run_all

async def run_all() -> tuple[Any, dict[str, Any]]

source

run_liveness

async def run_liveness() -> tuple[Any, dict[str, Any]]

source

run_readiness

async def run_readiness() -> tuple[Any, dict[str, Any]]

source

run_startup

async def run_startup() -> tuple[Any, dict[str, Any]]

source

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Satisfy HealthCheckProtocol.

---

HealthCheckResult

source

Result of a health check.

to_dict

def to_dict() -> dict[str, Any]

source

Convert to dictionary format.

is_healthy

def is_healthy() -> bool

source

Check if status is healthy.

is_degraded

def is_degraded() -> bool

source

Check if status is degraded.

---

HealthCheckRunHook

source

Payload fired after a health check probe completes.

Attributes: check_name: Name of the health check that ran. healthy: True if the check passed.

---

HealthChecker

source

Aggregates and runs named health checks with optional per-check timeouts.

Checks can be tagged with a HealthCheckCategory (LIVENESS, READINESS, or STARTUP) and run selectively via run_liveness, run_readiness, or run_startup. Methods that omit a category default to READINESS.

__init__

def __init__() -> None

source

add

def add(
    name: str,
    check: HealthCheckProtocol | Callable[[], Any],
    *,
    timeout: float | None = None,
    category: HealthCheckCategory = HealthCheckCategory.READINESS
) -> None

source

Add a named health check with an optional timeout and category.

Parameters

Parameter	Type	Description
`name`	str	Unique identifier for this check.
`check`	HealthCheckProtocol | Callable[[], Any]	A HealthCheckProtocol instance or a callable returning a health indicator (bool, dict, or HealthCheckResult).
`timeout`	float | None	Per-check timeout in seconds. If the check exceeds this duration it is reported as UNHEALTHY.
`category`	HealthCheckCategory	Whether this check is a liveness, readiness, or startup check. Defaults to ``READINESS``.

remove

def remove(name: str) -> None

source

Remove a named health check.

Parameters

Parameter	Type	Description
`name`	str	The check name to remove.

Raises

Exception	Description
KeyError	If no check with this name is registered.

has

def has(name: str) -> bool

source

Check whether a health check with the given name is registered.

Parameters

Parameter	Type	Description
`name`	str	The check name.

Returns

Type	Description
bool	True if the check exists.

check_names

property check_names() -> list[str]

source

Sorted list of all registered health check names.

Returns

Type	Description
list[str]	A sorted list of check name strings.

register

def register(check: Callable[[], Any]) -> None

source

Register a callable decorated with health_checker.

The check name is read from the _health_check_name attribute stamped by the health_checker decorator.

Parameters

Parameter	Type	Description
`check`	Callable[[], Any]	A decorated health check callable.

Raises

Exception	Description
ValueError	If *check* was not decorated with health_checker.

run_all

async def run_all() -> dict[str, HealthCheckResult]

source

Run all health checks and return unified results.

Per-check timeouts are enforced when configured. Timed-out checks are reported as UNHEALTHY.

Returns

Type	Description
dict[str, HealthCheckResult]	A dict mapping check names to their ``HealthCheckResult``.

aggregate_status

def aggregate_status(results: dict[str, HealthCheckResult]) -> HealthStatus

source

Determine the overall health status from individual check results.

Returns HEALTHY only if every check is HEALTHY. Returns DEGRADED if any check is DEGRADED but none are UNHEALTHY. Returns UNHEALTHY otherwise.

Parameters

Parameter	Type	Description
`results`	dict[str, HealthCheckResult]	The dict returned by run_all.

Returns

Type	Description
HealthStatus	The aggregate ``HealthStatus``.

run_all_with_summary

async def run_all_with_summary() -> tuple[HealthStatus, dict[str, HealthCheckResult]]

source

Run all health checks and return the aggregate status together with details.

Convenience wrapper that calls run_all and aggregate_status in one step.

Returns

Type	Description
tuple[HealthStatus, dict[str, HealthCheckResult]]	A tuple of ``(aggregate_status, per_check_results)``.

run_liveness

async def run_liveness() -> tuple[HealthStatus, dict[str, HealthCheckResult]]

source

Run only LIVENESS checks and return aggregate status and details.

Returns

Type	Description
tuple[HealthStatus, dict[str, HealthCheckResult]]	A tuple of ``(aggregate_status, per_check_results)``.

run_readiness

async def run_readiness() -> tuple[HealthStatus, dict[str, HealthCheckResult]]

source

Run only READINESS checks and return aggregate status and details.

Returns

Type	Description
tuple[HealthStatus, dict[str, HealthCheckResult]]	A tuple of ``(aggregate_status, per_check_results)``.

run_startup

async def run_startup() -> tuple[HealthStatus, dict[str, HealthCheckResult]]

source

Run only STARTUP checks and return aggregate status and details.

Returns

Type	Description
tuple[HealthStatus, dict[str, HealthCheckResult]]	A tuple of ``(aggregate_status, per_check_results)``.

---

HealthCheckerRegistry

source

Singleton-style registry for named CachedHealthChecker instances.

Manages the lifecycle of zero or more named checkers. The registry is registered as a container singleton by MonitorProvider; application code should resolve it via constructor injection rather than instantiating it directly.

Lifecycle

# At application startup — wired by MonitorProvider automatically.
registry = HealthCheckerRegistry()
checker = registry.get_or_create("api", db_provider=db)

# At application shutdown — MonitorProvider calls this.
await registry.cleanup()

# At application startup — wired by MonitorProvider automatically.
registry = HealthCheckerRegistry()
checker = registry.get_or_create("api", db_provider=db)

# At application shutdown — MonitorProvider calls this.
await registry.cleanup()

__init__

def __init__() -> None

source

get_or_create

def get_or_create(
    key: str = 'default',
    db_provider: Any = None,
    cache_ttl: float = 5.0,
    check_timeout: float = 2.0
) -> CachedHealthChecker

source

Return the named checker, creating it if it does not yet exist.

Subsequent calls with the same key always return the same instance regardless of db_provider, cache_ttl, or check_timeout — those arguments are ignored after the first call for a given key.

Parameters

Parameter	Type	Description
`key`	str	Logical name for this checker. Use distinct keys to manage independent groups of health checks (e.g. ``"api"`` vs ``"worker"``).
`db_provider`	Any	Optional database provider injected into the checker for database connectivity verification.
`cache_ttl`	float	Seconds to cache the last health-check result before re-running the underlying checks. Defaults to ``5.0``.
`check_timeout`	float	Per-check execution timeout in seconds. Defaults to ``2.0``.

Returns

Type	Description
CachedHealthChecker	The CachedHealthChecker registered under *key*.

get_checker

def get_checker(key: str = 'default') -> CachedHealthChecker | None

source

Return the checker registered under key, or None if absent.

Parameters

Parameter	Type	Description
`key`	str	The logical name used when the checker was created.

Returns

Type	Description
CachedHealthChecker | None	The CachedHealthChecker or ``None``.

list_checkers

def list_checkers() -> dict[str, CachedHealthChecker]

source

Return a snapshot of all registered checkers keyed by name.

Returns

Type	Description
dict[str, CachedHealthChecker]	A shallow copy of the internal registry mapping.

cleanup

async def cleanup() -> None

source

Stop background refresh tasks for every checker and clear the registry.

Called automatically by MonitorProvider during application shutdown. It is safe to call multiple times.

---

HealthStatus

source

Unified health status.

---

Histogram

source

Histogram for measuring distributions.

__init__

def __init__(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None,
    buckets: list[float] | None = None
)

source

name

property name() -> str

source

description

property description() -> str

source

record

def record(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

observe

def observe(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

Alias for record — delegates to the canonical implementation.

get_observations

def get_observations() -> list[float]

source

get_bucket_counts

def get_bucket_counts() -> dict[float, int]

source

reset

def reset() -> None

source

Clear all observations and values for metric rotation.

---

InMemoryTraceProvider

source

In-memory trace provider using Tracer.

__init__

def __init__(
    service_name: str = 'lexigram-service',
    max_spans: int = DEFAULT_MAX_SPANS,
    exporter: SpanExporter | None = None
) -> None

source

create_span

def create_span(
    name: str,
    parent: Span | None = None
) -> Span

source

start_span

def start_span(
    name: str,
    attributes: dict[str, Any] | None = None,
    context: Any | None = None
) -> Span

source

inject_context

def inject_context(
    carrier: dict[str, str],
    context: Any | None = None
) -> None

source

extract_context

def extract_context(carrier: dict[str, str]) -> Any | None

source

get_current_span

def get_current_span() -> Span | None

source

set_current_span

def set_current_span(span: Span | None) -> None

source

get_all_spans

def get_all_spans() -> list[Span]

source

---

LoggingAlertDispatcher

source

Dispatches alerts to the structured logger.

Implements AlertDispatcherProtocol.

Low and medium-severity alerts are logged at WARNING level; high and critical alerts are logged at ERROR level.

This implementation is always available with no external dependencies and is registered by MonitorProvider as the default AlertDispatcherProtocol binding when no other dispatcher is configured.

send_alert

async def send_alert(
    title: str,
    message: str,
    severity: str,
    context: dict[str, Any] | None = None
) -> None

source

Dispatch a free-form operational alert to the structured logger.

Parameters

Parameter	Type	Description
`title`	str	Short, human-readable alert title.
`message`	str	Detailed alert message.
`severity`	str	Severity level string, e.g. ``"low"``, ``"high"``, ``"critical"``.
`context`	dict[str, Any] | None	Optional free-form mapping of additional metadata included as structured key-value pairs on the log record.

send_metric_alert

async def send_metric_alert(
    metric_name: str,
    current_value: float,
    threshold: float,
    context: dict[str, Any] | None = None
) -> None

source

Dispatch a metric-threshold alert to the structured logger.

The alert is always logged at WARNING level unless the breach exceeds the threshold by more than 100 %, in which case ERROR is used.

Parameters

Parameter	Type	Description
`metric_name`	str	Name of the metric that breached its threshold.
`current_value`	float	Observed metric value at the time of the alert.
`threshold`	float	The configured threshold that was exceeded.
`context`	dict[str, Any] | None	Optional free-form mapping of additional metadata.

---

LoggingConfig

source

Configuration for structured logging.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

Attributes: enabled: Whether structured logging is enabled. level: Default log level. format: Log format (json, text). include_trace_context: Include trace context in logs. redact_fields: Fields to redact from logs.

validate_level

def validate_level(
    cls,
    v: str
) -> str

source

Validate log level.

---

MetricProxy

source

Proxy for a specific metric that provides a stateful API.

Bridges the gap between the MetricsCollectorProtocol (which is name-based) and consumers who expect a persistent metric object.

__init__

def __init__(
    name: str,
    collector: MetricsCollectorProtocol,
    metrics_type: str
) -> None

source

name

property name() -> str

source

MetricProtocol name.

increment

def increment(amount: float = 1.0) -> None

source

Increment the counter.

record

def record(value: float) -> None

source

Record a histogram value.

value

property value() -> float

source

Current counter value (tracked locally).

---

MetricRecordedHook

source

Payload fired when a metric data point is recorded.

Attributes: metric_name: Name of the metric being recorded. value: Numeric value of the data point.

---

MetricValue

source

Represents a metric measurement.

---

MetricsCollectorProtocol

source

Central metrics collection and reporting.

__init__

def __init__() -> None

source

register_metric

def register_metric(metric: MetricProtocol) -> None

source

Register an existing metric instrument.

create_counter

def create_counter(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
) -> Counter

source

create_gauge

def create_gauge(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
) -> Gauge

source

create_summary

def create_summary(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
) -> Summary

source

create_histogram

def create_histogram(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None,
    buckets: list[float] | None = None
) -> Histogram

source

get_metric

def get_metric(name: str) -> MetricProtocol | None

source

get_all_metrics

def get_all_metrics() -> dict[str, MetricProtocol]

source

clear

def clear() -> None

source

increment

def increment(
    name: str,
    value: float = 1.0,
    tags: dict[str, str] | None = None
) -> None

source

Increment a counter metric.

gauge

def gauge(
    name: str,
    value: float,
    tags: dict[str, str] | None = None
) -> None

source

Set a gauge metric.

histogram

def histogram(
    name: str,
    value: float,
    tags: dict[str, str] | None = None
) -> None

source

Record a histogram value.

---

MetricsConfig

source

Configuration for metrics collection.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

Attributes: enabled: Whether metrics collection is enabled. prefix: Prefix for all metric names. default_labels: Default labels to add to all metrics. histogram_buckets: Default bucket boundaries for histograms. collection_interval: Interval for periodic metrics collection (seconds).

make_metric_name

def make_metric_name(name: str) -> str

source

Create a prefixed metric name.

Parameters

Parameter	Type	Description
`name`	str	The metric name.

Returns

Type	Description
str	Prefixed metric name.

---

MetricsExporterRegistry

source

Central registry for metrics exporters.

Manages a collection of handlers for creating metrics exporters. Handlers are checked in order until one is found that can handle the exporter type.

Example

registry = MetricsExporterRegistry()
registry.register(CustomMetricsExporterHandler())
exporter = registry.create_exporter(config)

__init__

def __init__() -> None

source

Initialize the registry without any handlers.

with_defaults

def with_defaults(cls) -> MetricsExporterRegistry

source

Create a registry pre-populated with the built-in default handlers.

Returns

Type	Description
MetricsExporterRegistry	A new registry instance with all default metrics exporter handlers registered.

register

def register(handler: MetricsExporterHandler) -> None

source

Register a new metrics exporter handler.

Parameters

Parameter	Type	Description
`handler`	MetricsExporterHandler	The handler to register. Added at the beginning of the handler list.

create_exporter

def create_exporter(exp_config: Any) -> Any

source

Create an exporter for the given config.

Parameters

Parameter	Type	Description
`exp_config`	Any	Configuration object with a 'type' attribute.

Returns

Type	Description
Any	An exporter instance, or None if no handler can handle the type.

---

MonitorConfig

source

Hierarchical root configuration for Lexigram Monitor.

Attributes: enabled: Whether monitoring is enabled name: Configuration name (default: “monitor”) backend_type: Monitoring backend type metrics: Metrics configuration tracing: Tracing configuration health: Health check configuration logging: Logging configuration opentelemetry: OpenTelemetry configuration prometheus: Prometheus configuration environment: Environment name debug: Debug mode

is_production

property is_production() -> bool

source

Check if running in production environment.

is_development

property is_development() -> bool

source

Check if running in development environment.

is_test

property is_test() -> bool

source

Check if running in test environment.

validate_for_environment

def validate_for_environment(env: Environment | None = None) -> list[ConfigIssue]

source

Validate monitoring configuration for the given environment.

Parameters

Parameter	Type	Description
`env`	Environment | None	Target environment; resolved from ``LEX_ENV`` when ``None``.

Returns

Type	Description
list[ConfigIssue]	List of ConfigIssue instances.

get_backend_config

def get_backend_config() -> OpenTelemetryConfig | PrometheusConfig | None

source

Get the configuration for the selected backend.

Returns

Type	Description
OpenTelemetryConfig | PrometheusConfig | None	Backend-specific configuration or None for memory backend.

make_exporter

def make_exporter() -> Any | None

source

Construct an optional metrics exporter appropriate for the backend.

Returns

Type	Description
Any | None	MetricsExporter instance or None if no exporter is available.

---

MonitorModule

source

Metrics collection, distributed tracing, health checks, and profiling.

Call configure to configure the monitoring subsystem.

Usage (no-op / development)

from lexigram.monitor.backends.noop import NoopMetricsBackend

@module(
    imports=[MonitorModule.configure(backend=NoopMetricsBackend())]
)
class AppModule(Module):
    pass

from lexigram.monitor.backends.noop import NoopMetricsBackend

@module(
    imports=[MonitorModule.configure(backend=NoopMetricsBackend())]
)
class AppModule(Module):
    pass

Usage (Prometheus)

from lexigram.monitor.backends.exporters.prometheus import PrometheusMetricsExporter
from lexigram.monitor.backends.prometheus import PrometheusBackend

@module(
    imports=[MonitorModule.configure(backend=PrometheusBackend())]
)
class AppModule(Module):
    pass

from lexigram.monitor.backends.exporters.prometheus import PrometheusMetricsExporter
from lexigram.monitor.backends.prometheus import PrometheusBackend

@module(
    imports=[MonitorModule.configure(backend=PrometheusBackend())]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    backend: Any = None,
    config: Any | None = None
) -> DynamicModule

source

Create a MonitorModule with explicit configuration.

Parameters

Parameter	Type	Description
`backend`	Any	``MetricsBackendProtocol`` implementation. Defaults to ``NoOpMetricsBackend`` when omitted.
`config`	Any | None	``MonitorConfig`` for advanced tracing and profiling settings.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return a no-op MonitorModule for unit testing.

Registers a NoOpMetricsBackend that discards all metrics. No external telemetry systems are connected.

Parameters

Parameter	Type	Description
`config`	Any	Optional test configuration override.

Returns

Type	Description
DynamicModule	A DynamicModule with noop metrics and tracing.

---

MonitorProvider

source

Monitoring and observability provider for Lexigram Framework

__init__

def __init__(
    backend: MonitoringBackend,
    exporter: Any | None = None,
    config: Any | None = None
)

source

from_config

def from_config(
    cls,
    config: MonitorConfig,
    **context: Any
) -> MonitorProvider

source

Create a MonitorProvider from config.

Delegates to the create_provider_from_config factory.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register monitoring services with the container.

Binds all monitoring singletons into the DI container. DB-backed exporter wiring is deferred to boot() where the full container graph is available.

boot

async def boot(container: BootContainerProtocol) -> None

source

Start the monitoring provider and wire the observability facade.

After all providers are registered the container graph is complete, so this is the correct place to resolve optional cross-provider dependencies such as the database exporter.

shutdown

async def shutdown() -> None

source

Shutdown the monitoring provider

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check monitoring provider health

record_request

def record_request(
    method: str,
    path: str,
    duration: float,
    status_code: int
) -> None

source

Record an HTTP request

record_connection_change

def record_connection_change(delta: int) -> None

source

Record connection count change

create_counter

def create_counter(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
)

source

Create a counter metric

create_gauge

def create_gauge(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
)

source

Create a gauge metric

create_histogram

def create_histogram(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None,
    buckets: list[float] | None = None
)

source

Create a histogram metric

---

NoOpHealthCheckRegistry

source

No-op registry for categorised health checks.

add

def add(
    name: str,
    check: Callable[[], Any],
    *,
    timeout: float | None = None,
    critical: bool = True,
    category: Any = None
) -> None

source

Ignore health check registration.

run_all

async def run_all() -> tuple[Any, dict[str, Any]]

source

Return an empty aggregate result.

run_liveness

async def run_liveness() -> tuple[Any, dict[str, Any]]

source

Return an empty liveness result.

run_readiness

async def run_readiness() -> tuple[Any, dict[str, Any]]

source

Return an empty readiness result.

run_startup

async def run_startup() -> tuple[Any, dict[str, Any]]

source

Return an empty startup result.

health_check

async def health_check() -> HealthCheckResult

source

Report the no-op registry as healthy.

---

NoOpMetricsCollector

source

Metrics collector that drops all observations.

__init__

def __init__() -> None

source

name

property name() -> str

source

Return a placeholder metric name.

description

property description() -> str

source

Return a placeholder metric description.

register_metric

def register_metric(metric: MetricProtocol) -> None

source

Ignore pre-created metrics.

increment

def increment(
    name: str,
    value: float = 1.0,
    tags: dict[str, str] | None = None
) -> None

source

Ignore counter increments.

gauge

def gauge(
    name: str,
    value: float,
    tags: dict[str, str] | None = None
) -> None

source

Ignore gauge values.

histogram

def histogram(
    name: str,
    value: float,
    tags: dict[str, str] | None = None
) -> None

source

Ignore histogram samples.

create_counter

def create_counter(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
) -> NoOpMetricsCollector

source

Return the collector itself.

create_gauge

def create_gauge(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
) -> NoOpMetricsCollector

source

Return the collector itself.

create_histogram

def create_histogram(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None,
    buckets: list[float] | None = None
) -> NoOpMetricsCollector

source

Return the collector itself.

record

def record(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

Ignore recorded metric values.

---

NoOpSpan

source

No-op span that silently drops all tracing operations.

__init__

def __init__(name: str = '') -> None

source

set_attribute

def set_attribute(
    key: str,
    value: Any
) -> None

source

Store span attributes for introspection in tests.

add_event

def add_event(
    name: str,
    attributes: dict[str, Any] | None = None
) -> None

source

Ignore span events.

record_exception

def record_exception(exception: Exception) -> None

source

Ignore recorded exceptions.

set_status

def set_status(status: str) -> None

source

Ignore span status updates.

end

def end() -> None

source

Mark the span as ended without side effects.

---

NoOpTracer

source

No-op tracer that returns NoOpSpan instances.

start_span

def start_span(
    name: str,
    attributes: dict[str, Any] | None = None,
    context: Any | None = None
) -> NoOpSpan

source

Start a no-op span.

get_current_span

def get_current_span() -> SpanProtocol | None

source

Return no active span.

inject_context

def inject_context(
    carrier: dict[str, str],
    context: Any | None = None
) -> None

source

Leave the carrier unchanged.

extract_context

def extract_context(carrier: dict[str, str]) -> Any | None

source

Return no extracted context.

---

OTLPMetricsExporterHandler

source

Handler for OTLP metrics exporter.

Creates an OTLPMetricExporter for sending metrics to an OTLP-compatible backend.

can_handle

def can_handle(exporter_type: str) -> bool

source

Check if this handler can handle the exporter type.

Parameters

Parameter	Type	Description
`exporter_type`	str	The type of exporter to check.

Returns

Type	Description
bool	True if exporter_type is "otlp", False otherwise.

create_exporter

def create_exporter(exp_config: Any) -> Any

source

Create an OTLP metrics exporter instance.

Parameters

Parameter	Type	Description
`exp_config`	Any	Configuration with endpoint and headers.

Returns

Type	Description
Any	An OTLPMetricExporter instance.

---

OTLPTracingExporterHandler

source

Handler for OTLP tracing exporter.

Creates an OTLPSpanExporter for sending traces to an OTLP-compatible backend.

can_handle

def can_handle(exporter_type: str) -> bool

source

Check if this handler can handle the exporter type.

Parameters

Parameter	Type	Description
`exporter_type`	str	The type of exporter to check.

Returns

Type	Description
bool	True if exporter_type is "otlp", False otherwise.

create_exporter

def create_exporter(exp_config: Any) -> Any

source

Create an OTLP tracing exporter instance.

Parameters

Parameter	Type	Description
`exp_config`	Any	Configuration with endpoint and headers.

Returns

Type	Description
Any	An OTLPSpanExporter instance.

---

OTelMiddleware

source

Middleware for OpenTelemetry tracing and metrics in Lexigram-Web.

Provides automatic tracing and metrics collection for HTTP requests. Gracefully degrades when OpenTelemetry is not available or has import issues.

Features:

• Automatic span creation for each HTTP request
• Extracts parent trace context from incoming headers
• Records request count and duration metrics
• Captures HTTP status codes and error conditions

Example

from lexigram.monitor import OTelMiddleware

app = OTelMiddleware(my_asgi_app)

__init__

def __init__(app: ASGIApp)

source

Initialize the OTel middleware.

Parameters

Parameter	Type	Description
`app`	ASGIApp	The ASGI application to wrap.

---

ObservabilityProvider

source

Provides observability: metrics, tracing, and health checks.

Registers lightweight no-op stubs for every observability contract so that application code can always resolve MetricsCollectorProtocol, TracerProtocol, and HealthCheckRegistryProtocol via DI — even when lexigram-monitor is not installed.

When lexigram-monitor is installed, its MonitorProvider runs at a higher priority and overrides these registrations with the full implementations. This provider must not import from lexigram.monitor directly; doing so would violate the core ↔ extension hierarchy.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register no-op observability stubs into the container.

lexigram-monitor’s own provider will override these at its higher priority if the package is present.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

No boot-time work required for the observability module.

shutdown

async def shutdown() -> None

source

No resources to release for the observability module.

---

ObservabilityService

source

Unified service for observability (traces, metrics, health).

Automatically detects if the lexigram-monitor extension is available and delegates to it. Otherwise, uses NoOp implementations that still allow application code to instrument without errors.

Parameters

Parameter	Type	Description
`tracer`		Optional tracer backend. If ``None``, uses NoOpTracer.
`meter`		Optional meter backend. If ``None``, uses NoOpMetricsCollector.

__init__

def __init__(
    tracer: TracerProtocol | None = None,
    meter: MetricsCollectorProtocol | None = None
) -> None

source

register_metric

def register_metric(metric: MetricProtocol) -> None

source

Register an existing metric instrument.

Parameters

Parameter	Type	Description
`metric`	MetricProtocol	Pre-defined metric instance.

trace

def trace(
    name: str,
    **attributes: Any
) -> Iterator[SpanProtocol]

source

Create a tracing span.

If a real tracer is configured, delegates to its start_span method. Otherwise yields a NoOpSpan.

Parameters

Parameter	Type	Description
`name`	str	Span name. **attributes: Initial span attributes.

Yields

Type	Description
Iterator[SpanProtocol]	A span object with ``set_attribute()`` and ``add_event()`` methods.

counter

def counter(name: str) -> MetricProxy

source

Get or create a counter metric.

Parameters

Parameter	Type	Description
`name`	str	Counter name.

Returns

Type	Description
MetricProxy	A counter with ``increment()`` method.

histogram

def histogram(name: str) -> MetricProxy

source

Get or create a histogram metric.

Parameters

Parameter	Type	Description
`name`	str	Histogram name.

Returns

Type	Description
MetricProxy	A histogram with ``record()`` method.

timed

def timed(metric_name: str) -> Iterator[None]

source

Context manager that records execution duration as a histogram.

Parameters

Parameter	Type	Description
`metric_name`	str	The histogram metric name.

Yields

Type	Description
Iterator[None]	None — timing is recorded on exit.

---

OpenTelemetryBackend

source

OpenTelemetry monitoring backend.

__init__

def __init__(
    service_name: str = 'lexigram-app',
    endpoint: str | None = None,
    config: Any | None = None,
    tracing_exporter_registry: TracingExporterRegistry | None = None,
    metrics_exporter_registry: MetricsExporterRegistry | None = None
)

source

initialize

async def initialize() -> None

source

Initialize OpenTelemetry.

shutdown

async def shutdown() -> None

source

record_metric

def record_metric(
    name: str,
    value: Any,
    metric_type: str,
    labels: dict[str, str] | None = None
) -> None

source

create_span

def create_span(
    name: str,
    parent_context: SpanContext | None = None
) -> Span

source

---

OpenTelemetryConfig

source

Configuration for OpenTelemetry backend.

Attributes: endpoint: OTLP endpoint URL. headers: Headers to send with OTLP requests. insecure: Use insecure connection (no TLS). timeout: Export timeout in seconds. compression: Compression type (none, gzip). batch_size: Batch size for exports. export_interval: Export interval in seconds.

validate_compression

def validate_compression(
    cls,
    v: str
) -> str

source

Validate compression type.

---

PerformanceMetrics

source

Aggregated performance metrics.

duration

property duration() -> float | None

source

samples_per_second

property samples_per_second() -> float

source

---

PerformanceMonitor

source

Async performance monitor with periodic CPU/memory sampling.

Collects snapshots at config.sampling_interval intervals while active. Use start_monitoring / stop_monitoring or the async monitor_context context manager to bracket monitored regions.

__init__

def __init__(config: PerformanceMonitorConfig | None = None) -> None

source

Initialise monitor with optional configuration.

Parameters

Parameter	Type	Description
`config`	PerformanceMonitorConfig | None	Performance monitoring configuration (defaults apply).

state

property state() -> PerformanceMonitorState

source

Return the current monitoring state.

metrics

property metrics() -> PerformanceMetrics

source

Return the accumulated performance metrics.

start_monitoring

async def start_monitoring() -> None

source

Start periodic metrics collection.

Raises

Exception	Description
PerformanceMonitorError	If monitoring is already active.

stop_monitoring

async def stop_monitoring() -> None

source

Stop metrics collection and finalise metrics.

monitor_context

async def monitor_context() -> AsyncGenerator[PerformanceMonitor, None]

source

Async context manager that starts and stops monitoring automatically.

Yields

Type	Description
AsyncGenerator[PerformanceMonitor, None]	This monitor instance.

profile_function

async def profile_function(
    func: Any,
    *args: Any,
    **kwargs: Any
) -> FunctionProfileResult

source

Profile a synchronous or asynchronous function.

Parameters

Parameter	Type	Description
`func`	Any	Callable to profile (sync or async). *args: Positional arguments for *func*. **kwargs: Keyword arguments for *func*.

Returns

Type	Description
FunctionProfileResult	FunctionProfileResult with timing and memory data.

---

PerformanceMonitorConfig

source

Configuration for performance monitoring.

---

PerformanceMonitorState

source

State of the performance monitor.

---

PerformanceSnapshot

source

A single performance metrics snapshot.

---

PrometheusBackend

source

Prometheus monitoring backend.

__init__

def __init__(port: int = 8000)

source

Initialize Prometheus backend.

Parameters

Parameter	Type	Description
`port`	int	Port for metrics HTTP server

Raises

Exception	Description
BackendNotAvailableError	If prometheus-client is not installed

initialize

async def initialize() -> None

source

Initialize Prometheus metrics server.

If the configured port is already in use, log a warning and continue without failing the entire application startup.

shutdown

async def shutdown() -> None

source

Shutdown Prometheus server.

record_metric

def record_metric(
    name: str,
    value: Any,
    metric_type: str,
    labels: dict[str, str] | None = None
) -> None

source

Record a metric using Prometheus.

Parameters

Parameter	Type	Description
`name`	str	MetricProtocol name
`value`	Any	MetricProtocol value
`metric_type`	str	Type of metric (counter, gauge, histogram)
`labels`	dict[str, str] | None	Optional metric labels

create_span

def create_span(
    name: str,
    parent_context: SpanContext | None = None
) -> Span

source

Create a span (Prometheus doesn’t do tracing, so this is a no-op).

Parameters

Parameter	Type	Description
`name`	str	Span name
`parent_context`	SpanContext | None	Optional parent span context

Returns

Type	Description
Span	Dummy span instance

---

PrometheusConfig

source

Configuration for Prometheus backend.

Attributes: port: Port for metrics HTTP server. path: Path for metrics endpoint. enable_default_metrics: Enable default process metrics. pushgateway_url: Optional Pushgateway URL for push-based metrics. push_interval: Interval for pushing metrics to Pushgateway (seconds). store_in_db: Whether to persist metric observations to the database. metrics_table: Name of the metrics table to write samples to.

---

PrometheusMetricsExporter

source

Async-compatible adapter for prometheus_client.

Wraps prometheus_client.Counter, Gauge, and Histogram objects with a thread-safe, async-friendly API. Blocking calls are offloaded to a thread pool via asyncio.to_thread so the event loop is never stalled.

Label schemas for each metric name are locked-in on first use; subsequent calls must supply the same label keys. If a call omits a previously-seen label key the value defaults to an empty string so the series is still valid.

A Prometheus-client ASGI application (/metrics endpoint) is available via metrics_app. Mount it in your web application

# Inside a WebProvider or application startup
metrics_route = await container.resolve("prometheus_metrics_app")
app.mount("/metrics", metrics_route)

# Inside a WebProvider or application startup
metrics_route = await container.resolve("prometheus_metrics_app")
app.mount("/metrics", metrics_route)

Parameters

Parameter	Type	Description
`registry`		Optional custom ``CollectorRegistry``. Defaults to a fresh isolated registry so multiple exporters in the same process do not collide.

__init__

def __init__(registry: Any = None) -> None

source

metrics_app

property metrics_app() -> Any

source

ASGI application that exposes the Prometheus /metrics endpoint.

Returns

Type	Description
Any	An ASGI-compatible callable (from ``prometheus_client.make_asgi_app``).

Raises

Exception	Description
RuntimeError	If ``prometheus-client`` is not installed.

counter

async def counter(
    name: str,
    value: int,
    tags: dict[str, str] | None = None
) -> None

source

Increment a counter by value.

Parameters

Parameter	Type	Description
`name`	str	MetricProtocol name.
`value`	int	Amount to increment.
`tags`	dict[str, str] | None	Label key/value pairs.

gauge

async def gauge(
    name: str,
    value: float,
    tags: dict[str, str] | None = None
) -> None

source

Set a gauge to value.

Parameters

Parameter	Type	Description
`name`	str	MetricProtocol name.
`value`	float	New gauge value.
`tags`	dict[str, str] | None	Label key/value pairs.

histogram

async def histogram(
    name: str,
    value: float,
    tags: dict[str, str] | None = None
) -> None

source

Observe value on a histogram.

Parameters

Parameter	Type	Description
`name`	str	MetricProtocol name.
`value`	float	Observation value.
`tags`	dict[str, str] | None	Label key/value pairs.

flush

async def flush() -> None

source

No-op — prometheus_client pushes are pull-based.

---

PrometheusMiddleware

source

ASGI middleware that automatically collects HTTP request metrics.

Wraps any ASGI application and records per-request counters, duration histograms, and an active-request gauge. When prometheus_client is installed the metrics are exposed in the standard Prometheus text format at path (default /metrics); otherwise the framework’s MetricsCollectorProtocol is used so the application degrades gracefully without the optional dependency.

Parameters

Parameter	Type	Description
`path`		URL path at which to serve the Prometheus scrape endpoint. Defaults to ``"/metrics"``.
`metrics_collector`		Optional pre-configured MetricsCollectorProtocol. A new default collector is created when ``None`` is passed.

Example

from lexigram.monitor.middleware import PrometheusMiddleware

app = PrometheusMiddleware(my_asgi_app, path="/metrics")
# Mount *app* in your ASGI server of choice.

from lexigram.monitor.middleware import PrometheusMiddleware

app = PrometheusMiddleware(my_asgi_app, path="/metrics")
# Mount *app* in your ASGI server of choice.

__init__

def __init__(
    path: str = '/metrics',
    metrics_collector: MetricsCollectorProtocol | None = None
)

source

set_next_app

def set_next_app(app: Callable) -> None

source

next_app

async def next_app(
    scope: dict,
    receive: Callable,
    send: Callable
) -> None

source

---

SLO

source

Defines an acceptable performance target for a monitored metric.

An SLO specifies the maximum tolerable value for a percentile of a metric (e.g. p99 latency < 200 ms) and the error-budget burn-rate threshold above which an alert should be raised.

Attributes: name: Unique name for this SLO (e.g. "api.p99_latency"). metric: Name of the metric to evaluate (e.g. "http.request.duration"). percentile: Target percentile expressed as a fraction between 0.0 and 1.0 (e.g. 0.99 for p99). threshold_ms: Maximum acceptable measured value in milliseconds. window: Measurement window for rolling evaluation. burn_rate_threshold: Fire an alert when the measured/threshold ratio equals or exceeds this value. A value of 1.0 means “alert the moment the threshold is exceeded”. description: Optional human-readable description of the SLO.

---

SLOMonitor

source

Tracks SLO compliance and fires alerts on error-budget exhaustion.

Records metric samples in memory, then evaluates each registered SLO when evaluate is called. Violations are returned as a list of SLOViolation objects and also emitted as structured log warnings.

Example

from datetime import timedelta
from lexigram.monitor.slo import SLO, SLOMonitor

monitor = SLOMonitor()
monitor.register(SLO(
    name="api.p99_latency",
    metric="http.request.duration",
    percentile=0.99,
    threshold_ms=200.0,
    window=timedelta(hours=1),
))

monitor.record_sample("http.request.duration", 150.0)
monitor.record_sample("http.request.duration", 350.0)

violations = await monitor.evaluate()

from datetime import timedelta
from lexigram.monitor.slo import SLO, SLOMonitor

monitor = SLOMonitor()
monitor.register(SLO(
    name="api.p99_latency",
    metric="http.request.duration",
    percentile=0.99,
    threshold_ms=200.0,
    window=timedelta(hours=1),
))

monitor.record_sample("http.request.duration", 150.0)
monitor.record_sample("http.request.duration", 350.0)

violations = await monitor.evaluate()

__init__

def __init__(max_samples_per_metric: int = 10000) -> None

source

register

def register(slo: SLO) -> None

source

Register an SLO for monitoring.

Parameters

Parameter	Type	Description
`slo`	SLO	The SLO to track.

Raises

Exception	Description
ValueError	If an SLO with the same name is already registered.

record_sample

def record_sample(
    metric: str,
    value_ms: float
) -> None

source

Record a measurement sample for a metric.

Parameters

Parameter	Type	Description
`metric`	str	The metric name (matches registered SLO ``metric`` fields).
`value_ms`	float	The measurement value in milliseconds.

evaluate

async def evaluate() -> list[SLOViolation]

source

Evaluate all registered SLOs against recorded samples.

Returns

Type	Description
list[SLOViolation]	A list of SLOViolation instances for every SLO that is in breach. Returns an empty list when all SLOs are within bounds or when no samples have been recorded yet.

registered_slos

property registered_slos() -> dict[str, SLO]

source

Return registered SLOs keyed by name.

clear_samples

def clear_samples(metric: str | None = None) -> None

source

Remove recorded samples.

Parameters

Parameter	Type	Description
`metric`	str | None	If given, only clears samples for that metric. When ``None``, clears all recorded samples.

---

SLOViolation

source

Represents a detected SLO threshold breach.

Attributes: slo: The SLO that was violated. measured_value: The percentile value that exceeded the threshold (ms). burn_rate: ratio of measured_value to slo.threshold_ms at the time of detection. detected_at: UTC timestamp when the violation was detected. message: Human-readable summary, auto-generated when empty.

---

SamplerType

source

Supported tracing sampler types.

---

Span

source

Distributed tracing span.

set_attribute

def set_attribute(
    key: str,
    value: Any
) -> None

source

add_event

def add_event(
    name: str,
    attributes: dict[str, Any] | None = None
) -> None

source

set_status

def set_status(status: str) -> None

source

record_exception

def record_exception(exception: Exception) -> None

source

end

def end() -> None

source

get_duration

def get_duration() -> float | None

source

finish

def finish() -> None

source

Alias for end() for compatibility.

---

SpanContext

source

Span context for propagation.

to_traceparent

def to_traceparent() -> str

source

from_traceparent

def from_traceparent(
    cls,
    traceparent: str
) -> SpanContext | None

source

---

SpanKind

source

Span kind following OpenTelemetry specification.

---

SpanStatus

source

Span status following OpenTelemetry specification.

---

Summary

source

Summary metric for calculating quantiles.

__init__

def __init__(
    name: str,
    description: str = '',
    labels: dict[str, str] | None = None
)

source

name

property name() -> str

source

description

property description() -> str

source

record

def record(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

observe

def observe(
    value: float,
    labels: dict[str, str] | None = None
) -> None

source

Alias for record — delegates to the canonical implementation.

get_observations

def get_observations() -> list[float]

source

reset

def reset() -> None

source

Clear all observations and values for metric rotation.

---

Tracer

source

Distributed tracer logic.

__init__

def __init__(
    service_name: str,
    exporter: SpanExporter,
    max_spans: int | None = None
) -> None

source

start_span

def start_span(
    name: str,
    attributes: dict[str, Any] | None = None,
    context: Any | None = None,
    kind: SpanKind = SpanKind.INTERNAL
) -> Span

source

flush

def flush() -> None

source

get_all_spans

def get_all_spans() -> list[Span]

source

get_current_span

def get_current_span() -> Span | None

source

inject_context

def inject_context(
    carrier: dict[str, str],
    context: SpanContext | None = None
) -> None

source

Inject trace context into carrier dict.

extract_context

def extract_context(carrier: Mapping[str, str]) -> SpanContext | None

source

Extract trace context from carrier.

---

TracingConfig

source

Configuration for distributed tracing.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

Attributes: enabled: Whether tracing is enabled. service_name: Name of the service for traces. sampler_type: Type of sampling strategy. sample_rate: Sampling rate (0.0 to 1.0) for probability sampler. max_traces_per_second: Max traces/sec for rate limiting sampler. propagation_formats: Trace context propagation formats. max_attributes: Maximum attributes per span. max_events: Maximum events per span. max_links: Maximum links per span.

validate_sample_rate

def validate_sample_rate(
    cls,
    v: float
) -> float

source

Validate sample rate is between 0 and 1.

---

TracingExporterRegistry

source

Central registry for tracing exporters.

Manages a collection of handlers for creating tracing exporters. Handlers are checked in order until one is found that can handle the exporter type.

Example

registry = TracingExporterRegistry()
registry.register(CustomTracingExporterHandler())
exporter = registry.create_exporter(config)

__init__

def __init__() -> None

source

Initialize the registry without any handlers.

with_defaults

def with_defaults(cls) -> TracingExporterRegistry

source

Create a registry pre-populated with the built-in default handlers.

Returns

Type	Description
TracingExporterRegistry	A new registry instance with all default tracing exporter handlers registered.

register

def register(handler: TracingExporterHandler) -> None

source

Register a new tracing exporter handler.

Parameters

Parameter	Type	Description
`handler`	TracingExporterHandler	The handler to register. Added at the beginning of the handler list.

create_exporter

def create_exporter(exp_config: Any) -> Any

source

Create an exporter for the given config.

Parameters

Parameter	Type	Description
`exp_config`	Any	Configuration object with a 'type' attribute.

Returns

Type	Description
Any	An exporter instance, or None if no handler can handle the type.

---

Functions

get_performance_summary

get_performance_summary

async def get_performance_summary(metrics: PerformanceMetrics) -> dict[str, Any]

source

Return a dictionary summary of performance metrics.

Parameters

Parameter	Type	Description
`metrics`	PerformanceMetrics	Accumulated PerformanceMetrics to summarise.

Returns

Type	Description
dict[str, Any]	Dictionary containing duration, sample counts, CPU/memory averages, task counts, and a ``has_profile_data`` flag.

---

health_checker

health_checker

def health_checker(name: str) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]

source

Decorator that marks a callable as a named health check.

The decorated function can later be registered on a HealthChecker via HealthChecker.register.

Parameters

Parameter	Type	Description
`name`	str	Human-readable name for this check (e.g. ``"database"``).

Example

@health_checker("database")
async def check_db() -> bool:
    return await db.ping()

checker = HealthChecker()
checker.register(check_db)

@health_checker("database")
async def check_db() -> bool:
    return await db.ping()

checker = HealthChecker()
checker.register(check_db)

---

inject_trace_context

inject_trace_context

def inject_trace_context(carrier: dict[str, Any]) -> None

source

Inject current trace context into carrier for propagation.

Call this before sending a message to propagate trace context.

---

instrument_database

instrument_database

def instrument_database(provider: Any) -> None

source

Instrument a Lexigram DatabaseService with OpenTelemetry tracing.

Wraps the provider’s execute and execute_query methods to automatically create spans and record metrics for each database operation.

Features:

• Automatic span creation for each query
• Query duration tracking
• Error recording and status tracking
• Metrics for query count and duration

Parameters

Parameter	Type	Description
`provider`	Any	The database provider to instrument. Must have 'execute' and optionally 'execute_query' methods.

Example

from lexigram.monitor import instrument_database
from lexigram.sql import DatabaseService

provider = DatabaseService(config)
instrument_database(provider)
# Queries are now automatically traced

---

metered

metered

def metered(
    metric_name: str | None = None,
    *,
    service: ObservabilityService | None = None
) -> Callable[[F], F]

source

Decorator that records function execution time as a histogram metric.

Parameters

Parameter	Type	Description
`metric_name`	str | None	Explicit metric name. Defaults to ``{module}.{qualname}.duration``.
`facade`		Specific facade instance. Uses module default if ``None``.

Returns

Type	Description
Callable[[F], F]	Decorated function.

---

monitor

monitor

def monitor(
    name: str | None = None,
    *,
    log_args: bool = False
) -> Callable[[F], F]

source

Wrap a function with combined structured logging and wall-clock timing.

Convenience alternative to stacking @traced and @metered. Emits operation_start, operation_end, and operation_error log events with the elapsed duration in milliseconds. Works for both async and sync callables.

Parameters

Parameter	Type	Description
`name`	str | None	Metric/span name. Defaults to ``"{module}.{qualname}"``.
`log_args`	bool	When ``True``, the number of positional arguments is included in the log context. Defaults to ``False`` to avoid inadvertently logging sensitive data.

Returns

Type	Description
Callable[[F], F]	Decorator that wraps the target function with monitoring machinery.

Example

@monitor("user_service.create")
async def create_user(email: str) -> User:
    return await repo.save(User(email=email))

@monitor()
def compute_score(data: list[float]) -> float:
    return sum(data) / len(data)

@monitor("user_service.create")
async def create_user(email: str) -> User:
    return await repo.save(User(email=email))

@monitor()
def compute_score(data: list[float]) -> float:
    return sum(data) / len(data)

---

monitor_async_operation

monitor_async_operation

async def monitor_async_operation(config: PerformanceMonitorConfig | None = None) -> AsyncGenerator[PerformanceMonitor, None]

source

Async context manager that monitors a region of async code.

Creates a PerformanceMonitor, starts it before entering the body, and stops it on exit.

Parameters

Parameter	Type	Description
`config`	PerformanceMonitorConfig | None	Optional PerformanceMonitorConfig. Default config is used when not provided.

Yields

Type	Description
AsyncGenerator[PerformanceMonitor, None]	Running PerformanceMonitor instance.

Example

async with monitor_async_operation() as monitor:
await do_some_work()
print(monitor.metrics.duration)

---

profile_async_function

profile_async_function

async def profile_async_function(
    func: Any,
    *args: Any,
    **kwargs: Any
) -> tuple[Any, FunctionProfileResult]

source

Profile a single async (or sync) function call.

Parameters

Parameter	Type	Description
`func`	Any	Callable to profile. *args: Positional arguments forwarded to *func*. **kwargs: Keyword arguments forwarded to *func*.

Returns

Type	Description
tuple[Any, FunctionProfileResult]	A ``(result, profile)`` tuple where *result* is the return value of *func* and *profile* is a FunctionProfileResult.

Example

result, profile = await profile_async_function(my_async_fn, arg1)
print(profile.execution_time)

---

trace_consume

trace_consume

async def trace_consume(
    channel: str,
    message_type: str = 'unknown',
    carrier: dict[str, Any] | None = None,
    **attributes: Any
) -> AsyncGenerator[Span, None]

source

Context manager for tracing message consumption.

Usage

async with trace_consume(“notifications”, carrier=headers) as span: await process_message(…)

---

trace_publish

trace_publish

async def trace_publish(
    channel: str,
    message_type: str = 'email',
    recipient: str | None = None,
    **attributes: Any
) -> AsyncGenerator[Span, None]

source

Context manager for tracing message publishing.

Usage

async with trace_publish(“notifications”, “email”, “user@example.com”) as span: await send_email(…)

---

traced

traced

def traced(
    span_name: str | None = None,
    *,
    service: ObservabilityService | None = None
) -> Callable[[F], F]

source

Decorator that wraps a function in a tracing span.

Automatically captures function name, arguments, and any raised exceptions as span events.

Parameters

Parameter	Type	Description
`span_name`	str | None	Explicit span name. Defaults to the qualified function name.
`facade`		Specific facade instance. Uses module default if ``None``.

Returns

Type	Description
Callable[[F], F]	Decorated function.

---

Exceptions

BackendNotAvailableError

source

Raised when a required monitoring backend is not available.

---

InvalidMetricError

source

Raised when metric parameters are invalid.

---

MetricNotFoundError

source

Raised when a requested metric is not found.

---

MonitorError

source

Base exception for monitoring errors.

---

PerformanceMonitorError

source

Base exception for performance monitoring errors.

__init__

def __init__(
    message: str = 'Performance monitoring error',
    **kwargs
) -> None

source

---

SpanError

source

Base exception for span-related errors.

---