API Reference

Protocols

AlertDispatcherProtocol

source

Protocol for dispatching operational alerts.

Implementations route alerts to notification channels (logging, PagerDuty, Slack, etc.). quadkit-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.

---

ErrorTrackerProtocol

source

Minimal interface for capturing exceptions to an external service.

capture_exception

def capture_exception(exc: BaseException) -> None

source

Report an exception to the external error tracking service.

flush

def flush(timeout: float = 2.0) -> None

source

Flush buffered events to the external service.

Parameters

Parameter	Type	Description
timeout	float	Max seconds to wait for delivery.

---

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

get_series

def get_series() -> dict[SeriesKey, float]

source

Return the running total of every label set recorded so far.

Returns

Type	Description
dict[SeriesKey, float]	Mapping of canonical label key (sorted `(name, value)` pairs) to the amount recorded under that label set.

reset

def reset() -> None

source

Reset the counter to zero and clear the observation history.

---

ErrorTrackingConfig

source

External error tracking configuration (Sentry).

When dsn is unset (the default), error tracking is a no-op so the integration adds no overhead or network traffic unless explicitly enabled (env: QK_MONITOR__ERROR_TRACKING__DSN).

Attributes: dsn: Sentry DSN. Unset/empty disables error tracking entirely. environment: Environment tag attached to captured events. traces_sample_rate: Sampling rate for traces (0.0 to 1.0). send_default_pii: Whether to send default PII fields.

---

FunctionHealthCheck

source

Health check implemented as a function.

__init__

def __init__(
    name: str,
    func: Callable,
    critical: bool = True,
    timeout: float | None = None
)

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

get_series

def get_series() -> dict[SeriesKey, float]

source

Return the latest value of every label set recorded so far.

Returns

Type	Description
dict[SeriesKey, float]	Mapping of canonical label key (sorted `(name, value)` pairs) to the most recent value recorded under that label set.

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,
    timeout: float | None = None
)

source

Initialize the check.

Parameters

Parameter	Type	Description
name	str	Registered check name.
critical	bool	Whether an unhealthy result fails readiness.
timeout	float | None	Optional per-check seconds bound. Enforced by HealthCheckRegistry.run_all and friends; `None` leaves the check unbounded.

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.

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"`.
auth_token		Optional bearer token. When set, requests must carry `Authorization: Bearer or receive 401. Defaults to None` (endpoint open).

Example

from quadkit.monitor.middleware import HealthCheckProvider

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

from quadkit.monitor.middleware import HealthCheckProvider

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

__init__

def __init__(
    path: str = '/health',
    auth_token: str | None = None
) -> 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

run_check

async def run_check(name: str) -> dict[str, Any]

source

Run a single named check and return its raw result dict.

Parameters

Parameter	Type	Description
name	str	Name of the registered check.

Returns

Type	Description
dict[str, Any]	The check's result dict, or an `UNKNOWN` dict when the name is not registered in either probe list.

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(only: set[str] | None = None) -> 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.

Parameters

Parameter	Type	Description
only	set[str] | None	When given, run only the checks whose names are in this set (unknown names are ignored). `None` runs everything.

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_backend: 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_backend	Any	Optional cache backend injected into the checker for cache 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

get_series_observations

def get_series_observations() -> dict[SeriesKey, list[float]]

source

Return the observations recorded for every label set.

A Prometheus histogram is exposed per label set (one set of _bucket/_sum/_count series each); the flat _observations deque cannot be split that way after the fact, so observations are retained per label set here.

Returns

Type	Description
dict[SeriesKey, list[float]]	Mapping of canonical label key (sorted `(name, value)` pairs) to that label set's observations.

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 = 'quadkit-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

---

InfrastructureHealthProvider

source

Register database and cache readiness probes via Tier-0 protocols.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Register infra:db and infra:cache readiness checks.

---

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.

---

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.

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.

register

def register(handler: MetricsExporterHandler) -> None

source

Register a new metrics exporter handler.

Parameters

Parameter	Type	Description
handler	MetricsExporterHandler	The handler to register; inserted at the head of the 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 Quadkit 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 slo: SLO evaluation configuration error_tracking: External error tracking configuration opentelemetry: OpenTelemetry configuration prometheus: Prometheus configuration environment: Environment name debug: Debug mode

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 `self.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.

---

MonitorModule

source

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

Call configure to configure the monitoring subsystem.

Usage (no-op / development)

from quadkit.monitor.backends.noop import NoopMetricsBackend

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

from quadkit.monitor.backends.noop import NoopMetricsBackend

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

Usage (Prometheus)

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

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

from quadkit.monitor.backends.exporters.prometheus import PrometheusMetricsExporter
from quadkit.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,
    exporter: 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.
exporter	Any | None	Optional `MetricsExporter. A PrometheusMetricsExporter here is what registers the prometheus_metrics_app container key the web layer mounts at /metrics` — without it the documented mount has nothing to mount.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

with_slo

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

source

Return a MonitorModule that also exports SLO types.

Parameters

Parameter	Type	Description
backend	Any	`MetricsBackendProtocol` implementation.
config	Any | None	`MonitorConfig` for advanced settings.
exporter	Any | None	Optional `MetricsExporter` (see configure).

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 Quadkit 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.

---

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.

---

NullErrorTracker

source

No-op tracker used when no DSN is configured.

Keeps capture/flush calls safe so callers never need to branch on whether error tracking is active.

capture_exception

def capture_exception(exc: BaseException) -> None

source

Log the exception locally instead of reporting it externally.

flush

def flush(timeout: float = 2.0) -> None

source

Nothing to flush.

---

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 Quadkit-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 quadkit.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 quadkit-monitor is not installed.

When quadkit-monitor is installed, its MonitorProvider runs at a higher priority and overrides these registrations with the full implementations. This provider must not import from quadkit.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.

quadkit-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 quadkit-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 = 'quadkit-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
app		The inner ASGI application to forward non-scrape requests to. When `None`, non-metrics scopes are dropped (see set_next_app for post-construction wiring).
path		URL path at which to serve the Prometheus scrape endpoint. Defaults to `"/metrics"`.
metrics_collector		Optional pre-configured MetricsCollectorProtocol. When `None` the process-wide collector is used (the one the monitor provider binds into DI), so middleware- and service-recorded metrics land in the same place.
auth_token		Optional bearer token. When set, requests must carry `Authorization: Bearer or receive 401. Defaults to None` (endpoint open).

Example

from quadkit.monitor.middleware import PrometheusMiddleware

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

from quadkit.monitor.middleware import PrometheusMiddleware

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

__init__

def __init__(
    app: Callable[Ellipsis, Any] | None = None,
    path: str = '/metrics',
    metrics_collector: MetricsCollectorProtocol | None = None,
    auth_token: str | None = None
)

source

set_next_app

def set_next_app(app: Callable) -> None

source

Wire the inner ASGI application after construction.

---

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. tier: Alert routing tier. Defaults to P1_BUSINESS_HOURS. owner: Team or spec ID responsible for this SLO. runbook_url: URL pointing to the runbook for this 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.

When alert_dispatcher is provided, evaluate_and_dispatch sends each violation through the dispatcher with tier-aware context and respects suppression_window_seconds to avoid alert storms.

Example

from datetime import timedelta
from quadkit.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 quadkit.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,
    alert_dispatcher: AlertDispatcherProtocol | None = None,
    suppression_window_seconds: int = 300
) -> 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,
    *,
    at: datetime | None = None
) -> None

source

Record a measurement sample for a metric.

Samples carry a UTC timestamp so evaluate can restrict each SLO to its configured rolling window.

Parameters

Parameter	Type	Description
metric	str	The metric name (matches registered SLO `metric` fields).
value_ms	float	The measurement value in milliseconds.
at	datetime | None	Optional explicit sample timestamp (UTC). Defaults to "now"; accept an explicit value when replaying recorded data or in tests.

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.

evaluate_and_dispatch

async def evaluate_and_dispatch() -> list[SLOViolation]

source

Evaluate all SLOs and dispatch violations through the alert dispatcher.

Each violation is routed through self._alert_dispatcher with tier-aware context when available. Duplicate alerts for the same SLO are suppressed within suppression_window_seconds.

Returns

Type	Description
list[SLOViolation]	The full list of violations (same as evaluate).

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.

---

SentryErrorTracker

source

Sentry-backed tracker used when a DSN is configured.

__init__

def __init__(
    config: ErrorTrackingConfig,
    sentry: Any,
    dsn: str | None = None
) -> None

source

Initialize the Sentry SDK.

Parameters

Parameter	Type	Description
config	ErrorTrackingConfig	Error tracking config with a non-empty `dsn`.
sentry	Any	Imported `sentry_sdk` module.
dsn	str | None	Effective DSN to use; falls back to `config.dsn` when unset.

capture_exception

def capture_exception(exc: BaseException) -> None

source

Report the exception to Sentry.

flush

def flush(timeout: float = 2.0) -> None

source

Flush buffered events to Sentry.

Parameters

Parameter	Type	Description
timeout	float	Max seconds to wait for delivery.

---

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

get_series_observations

def get_series_observations() -> dict[SeriesKey, list[float]]

source

Return the observations recorded for every label set.

Returns

Type	Description
dict[SeriesKey, list[float]]	Mapping of canonical label key (sorted `(name, value)` pairs) to that label set's observations.

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.

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.

register

def register(handler: TracingExporterHandler) -> None

source

Register a new tracing exporter handler.

Parameters

Parameter	Type	Description
handler	TracingExporterHandler	The handler to register; inserted at the head of the 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

create_opentelemetry_provider

create_opentelemetry_provider

def create_opentelemetry_provider(
    service_name: str = 'quadkit-app',
    endpoint: str | None = None
) -> MonitorProvider

source

Create a MonitorProvider backed by OpenTelemetry.

Parameters

Parameter	Type	Description
service_name	str	Logical service name reported to the OTel collector.
endpoint	str | None	Optional OTLP exporter endpoint URL.

Returns

Type	Description
MonitorProvider	MonitorProvider configured with OpenTelemetryBackend.

---

create_prometheus_provider

create_prometheus_provider

def create_prometheus_provider(port: int = 8000) -> MonitorProvider

source

Create a MonitorProvider backed by Prometheus.

Parameters

Parameter	Type	Description
port	int	Port on which the Prometheus HTTP metrics endpoint is exposed.

Returns

Type	Description
MonitorProvider	MonitorProvider configured with PrometheusBackend.

---

create_provider_from_config

create_provider_from_config

def create_provider_from_config(config: MonitorConfig | dict[str, Any] | None) -> MonitorProvider

source

Create a MonitorProvider from a ``MonitorConfig`` object or mapping.

Selects the appropriate monitoring backend and, when available, wires an async MetricsExporter for tasks and background jobs.

Parameters

Parameter	Type	Description
config	MonitorConfig | dict[str, Any] | None	A `MonitorConfig instance, dict, or compatible mapping. A plain dict is coerced into MonitorConfig` automatically.

Returns

Type	Description
MonitorProvider	Fully configured MonitorProvider.

Raises

Exception	Description
ValueError	When *config* is `None`.

---

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 Quadkit 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 quadkit.monitor import instrument_database
from quadkit.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)

---

setup_error_tracking

setup_error_tracking

def setup_error_tracking(config: ErrorTrackingConfig) -> ErrorTrackerProtocol

source

Initialize error tracking from config; no-op when no DSN is set.

Parameters

Parameter	Type	Description
config	ErrorTrackingConfig	Error tracking configuration. When `dsn` is unset or blank, a NullErrorTracker is returned and no external connection is attempted.

Returns

Type	Description
ErrorTrackerProtocol	An active tracker when the DSN is set and `sentry-sdk` is installed, otherwise a no-op NullErrorTracker.

Example

tracker = setup_error_tracking(MonitorConfig().error_tracking)

---

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”, “[email protected]”) 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: Any
) -> None

source

---

SpanError

source

Base exception for span-related errors.

---