Observability

Observability and metrics for Confii.

This module provides metrics collection and observability features for monitoring configuration usage, access patterns, and performance.

class confii.observability.ConfigAccessMetric(key: str, access_count: int = 0, first_access: float | None = None, last_access: float | None = None, avg_access_time: float = 0.0, total_access_time: float = 0.0)

Bases: object

Represents a configuration access metric.

This dataclass tracks metrics for a single configuration key, including access frequency and timing information.

key

The configuration key being tracked

Type:

str

access_count

Total number of times this key has been accessed

Type:

int

first_access

Timestamp of the first access (seconds since epoch)

Type:

float | None

last_access

Timestamp of the most recent access (seconds since epoch)

Type:

float | None

avg_access_time

Average time taken to access this key (in seconds)

Type:

float

total_access_time

Total time spent accessing this key (in seconds)

Type:

float

__init__(key: str, access_count: int = 0, first_access: float | None = None, last_access: float | None = None, avg_access_time: float = 0.0, total_access_time: float = 0.0) → None

record_access(access_time: float = 0.0) → None

Record an access to this configuration key.

Parameters:

access_time – Time taken to access the key (in seconds)

class confii.observability.ConfigEventEmitter

Bases: object

Event emitter for configuration events.

This class provides a simple event emitter pattern for configuration events, allowing listeners to subscribe to configuration changes, reloads, and other events. Supports multiple listeners per event type.

_listeners

Dictionary mapping event names to lists of callback functions

Example

>>> from confii import Config
>>> config = Config(loaders=[YamlLoader("config.yaml")])
>>> emitter = config.enable_events()
>>>
>>> @emitter.on("reload")
... def handle_reload(new_config, duration):
...     print(f"Config reloaded in {duration:.3f}s")
>>>
>>> @emitter.on("change")
... def handle_change(old_config, new_config):
...     print("Configuration changed")

__init__() → None

Initialize the event emitter.

Creates a new event emitter with an empty listener registry.

emit(event: str, *args: Any, **kwargs: Any) → None

Emit an event to all listeners.

Parameters:

• event – Event name

• *args – Positional arguments for listeners

• **kwargs – Keyword arguments for listeners

Example

>>> emitter.emit("reload", config_dict)

off(event: str, callback: Callable) → None

Unregister an event listener.

Parameters:

• event – Event name

• callback – Callback function to remove

on(event: str, callback: Callable | None = None) → Callable

Register an event listener.

Can be used as a decorator or called directly:

# As decorator @emitter.on(“reload”) def handle_reload():

print(“Configuration reloaded”)

# Direct call emitter.on(“reload”, handle_reload)

Parameters:

• event – Event name (e.g., “reload”, “change”, “access”)

• callback – Callback function. If None, returns a decorator.

Returns:

The callback function (for decorator chaining)

class confii.observability.ConfigMetrics(total_keys: int = 0, accessed_keys: int = 0, reload_count: int = 0, last_reload: float | None = None, reload_durations: List[float] = <factory>, access_metrics: Dict[str, ~confii.observability.ConfigAccessMetric]=<factory>, change_count: int = 0, last_change: float | None = None)

Bases: object

Configuration metrics and statistics.

This dataclass aggregates metrics for configuration usage, including access patterns, reload frequency, and change tracking.

total_keys

Total number of keys in the configuration

Type:

int

accessed_keys

Number of unique keys that have been accessed

Type:

int

reload_count

Total number of times configuration has been reloaded

Type:

int

last_reload

Timestamp of the last reload (seconds since epoch)

Type:

float | None

reload_durations

List of reload durations (in seconds) for performance tracking

Type:

List[float]

access_metrics

Dictionary mapping key paths to ConfigAccessMetric instances

Type:

Dict[str, confii.observability.ConfigAccessMetric]

change_count

Total number of configuration changes recorded

Type:

int

last_change

Timestamp of the last configuration change (seconds since epoch)

Type:

float | None

MAX_RELOAD_DURATIONS = 1000

__init__(total_keys: int = 0, accessed_keys: int = 0, reload_count: int = 0, last_reload: float | None = None, reload_durations: List[float] = <factory>, access_metrics: Dict[str, ~confii.observability.ConfigAccessMetric]=<factory>, change_count: int = 0, last_change: float | None = None) → None

get_statistics() → Dict[str, Any]

Get metrics statistics.

Returns:

Dictionary with metrics statistics

record_access(key: str, access_time: float = 0.0) → None

Record a configuration key access.

Parameters:

• key – Configuration key accessed

• access_time – Time taken to access (in seconds)

record_change() → None

Record a configuration change.

record_reload(duration: float) → None

Record a configuration reload.

Parameters:

duration – Reload duration in seconds

class confii.observability.ConfigObserver

Bases: object

Observer for configuration access and changes.

This class tracks configuration usage patterns, access frequencies, and performance metrics for observability purposes. It can be enabled on a Config instance to automatically collect metrics during normal usage.

metrics

ConfigMetrics instance containing collected metrics

_enabled

Whether metrics collection is currently enabled

Example

>>> from confii import Config
>>> config = Config(loaders=[YamlLoader("config.yaml")])
>>> observer = config.enable_observability()
>>> # Use config normally - metrics are collected automatically
>>> host = config.database.host
>>> metrics = observer.get_statistics()
>>> print(f"Config accessed {metrics['accessed_keys']} times")

__init__() → None

Initialize the config observer.

Creates a new observer instance with metrics collection enabled by default.

disable() → None

Disable metrics collection.

When disabled, the observer stops tracking metrics but retains previously collected data.

enable() → None

Enable metrics collection.

When enabled, the observer will track all configuration access, reloads, and changes.

get_metrics() → ConfigMetrics

Get current metrics.

Returns:

ConfigMetrics instance

get_statistics() → Dict[str, Any]

Get metrics statistics.

Returns a dictionary containing aggregated statistics about configuration usage, including access patterns, reload performance, and top accessed keys.

Returns:

• total_keys: Total number of configuration keys

• accessed_keys: Number of unique keys accessed

• access_rate: Ratio of accessed keys to total keys

• reload_count: Number of times config was reloaded

• avg_reload_time: Average reload duration in seconds

• change_count: Number of configuration changes

• top_accessed_keys: List of most frequently accessed keys

Return type:

Dictionary with metrics statistics including

Example

>>> stats = observer.get_statistics()
>>> print(f"Top accessed key: {stats['top_accessed_keys'][0]['key']}")
>>> print(f"Average reload time: {stats['avg_reload_time']:.3f}s")

record_change() → None

Record a configuration change.

record_key_access(key: str, access_time: float = 0.0) → None

Record a configuration key access.

Parameters:

• key – Configuration key accessed

• access_time – Time taken to access (in seconds)

record_reload(duration: float) → None

Record a configuration reload.

Parameters:

duration – Reload duration in seconds

reset_metrics() → None

Reset all metrics to initial state.

Clears all collected metrics, including access history, reload durations, and change tracking. Useful for starting fresh metrics collection.

Example

>>> observer.reset_metrics()
>>> # Metrics are now reset to zero