Skip to content
v0.1.3

Observability Module API Reference

This section documents the internals of the observability module in Bijux CLI.

bijux_cli.infra.observability

Provides the concrete implementation of the observability and logging service.

This module defines the Observability class, which implements the ObservabilityProtocol. It serves as the primary interface for structured logging throughout the application, using structlog as its underlying engine. It can also be configured to forward log entries to a telemetry backend, unifying logging and event tracking.

Observability 

Observability(*, debug: bool = False)

Bases: ObservabilityProtocol

A structured logging service integrating structlog and telemetry.

This class wraps a structlog logger to produce structured log entries. If configured with a telemetry backend, it also forwards these events for analytics and monitoring.

Attributes:

  • _logger (FilteringBoundLogger) –

    The underlying structlog logger instance.

  • _telemetry (TelemetryProtocol) –

    The telemetry service for event forwarding. Defaults to a NullTelemetry instance that does nothing.

Initializes the observability service.

Parameters:

  • debug (bool, default: False ) –

    If True, configures the service for debug-level logging.

Source code in src/bijux_cli/infra/observability.py 
39
40
41
42
43
44
45
46
47
48
@inject
def __init__(self, *, debug: bool = False) -> None:
    """Initializes the observability service.

    Args:
        debug (bool): If True, configures the service for debug-level
            logging.
    """
    self._logger: FilteringBoundLogger = structlog.get_logger("bijux_cli")
    self._telemetry: TelemetryProtocol = NullTelemetry()

bind 

bind(**kwargs: Any) -> Self

Binds context key-value pairs to all subsequent log entries.

Parameters:

  • **kwargs (Any, default: {} ) –

    Context values to include in each log entry.

Returns:

  • Self ( Self ) –

    The service instance, allowing for method chaining.

Source code in src/bijux_cli/infra/observability.py 
86
87
88
89
90
91
92
93
94
95
96
def bind(self, **kwargs: Any) -> Self:
    """Binds context key-value pairs to all subsequent log entries.

    Args:
        **kwargs (Any): Context values to include in each log entry.

    Returns:
        Self: The service instance, allowing for method chaining.
    """
    self._logger = self._logger.bind(**kwargs)
    return self

close 

close() -> None

Logs the shutdown of the observability service.

Note

In this implementation, this method only logs a debug message and does not perform resource cleanup like flushing. Flushing is handled by the telemetry service's own lifecycle methods.

Source code in src/bijux_cli/infra/observability.py 
137
138
139
140
141
142
143
144
145
def close(self) -> None:
    """Logs the shutdown of the observability service.

    Note:
        In this implementation, this method only logs a debug message and
        does not perform resource cleanup like flushing. Flushing is
        handled by the telemetry service's own lifecycle methods.
    """
    self._logger.debug("Observability shutdown")

get_logger 

get_logger() -> FilteringBoundLogger

Retrieves the underlying structlog logger instance.

Returns:

  • FilteringBoundLogger ( FilteringBoundLogger ) –

    The structlog logger, which can be used directly if needed.

Source code in src/bijux_cli/infra/observability.py 
77
78
79
80
81
82
83
84
def get_logger(self) -> FilteringBoundLogger:
    """Retrieves the underlying `structlog` logger instance.

    Returns:
        FilteringBoundLogger: The `structlog` logger, which can be used
            directly if needed.
    """
    return self._logger

log 

log(
    level: str,
    msg: str,
    *,
    extra: dict[str, Any] | None = None,
) -> Self

Logs a structured message and emits a corresponding telemetry event.

Parameters:

  • level (str) –

    The severity level of the log (e.g., 'debug', 'info', 'warning', 'error', 'critical').

  • msg (str) –

    The log message.

  • extra (dict[str, Any] | None, default: None ) –

    Additional context to include in the log entry.

Returns:

  • Self ( Self ) –

    The service instance, allowing for method chaining.

Raises:

Source code in src/bijux_cli/infra/observability.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def log(
    self,
    level: str,
    msg: str,
    *,
    extra: dict[str, Any] | None = None,
) -> Self:
    """Logs a structured message and emits a corresponding telemetry event.

    Args:
        level (str): The severity level of the log (e.g., 'debug', 'info',
            'warning', 'error', 'critical').
        msg (str): The log message.
        extra (dict[str, Any] | None): Additional context to include in the
            log entry.

    Returns:
        Self: The service instance, allowing for method chaining.

    Raises:
        ServiceError: If `level` is not a valid log level name.
    """
    log_func = getattr(self._logger, level.lower(), None)
    if not callable(log_func):
        raise ServiceError(f"Invalid log level: {level}")

    log_context = extra or {}
    if log_context:
        log_func(msg, **log_context)
    else:
        log_func(msg)

    if not isinstance(self._telemetry, NullTelemetry):
        telemetry_payload = {"level": level, "message": msg}
        telemetry_payload.update(log_context)
        self._telemetry.event("LOG_EMITTED", telemetry_payload)

    return self

set_telemetry 

set_telemetry(telemetry: TelemetryProtocol) -> Self

Attaches a telemetry backend for forwarding log events.

This allows the service to be "upgraded" from a simple logger to a full observability tool after its initial creation.

Parameters:

Returns:

  • Self ( Self ) –

    The service instance, allowing for method chaining.

Source code in src/bijux_cli/infra/observability.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def set_telemetry(self, telemetry: TelemetryProtocol) -> Self:
    """Attaches a telemetry backend for forwarding log events.

    This allows the service to be "upgraded" from a simple logger to a full
    observability tool after its initial creation.

    Args:
        telemetry (TelemetryProtocol): The telemetry service to receive events.

    Returns:
        Self: The service instance, allowing for method chaining.
    """
    self._telemetry = telemetry
    return self

setup classmethod 

setup(*, debug: bool = False) -> Self

Instantiates and configures an Observability service.

Parameters:

  • debug (bool, default: False ) –

    If True, enables debug-level logging.

Returns:

  • Self ( Self ) –

    A new, configured Observability instance.

Source code in src/bijux_cli/infra/observability.py 
65
66
67
68
69
70
71
72
73
74
75
@classmethod
def setup(cls, *, debug: bool = False) -> Self:
    """Instantiates and configures an `Observability` service.

    Args:
        debug (bool): If True, enables debug-level logging.

    Returns:
        Self: A new, configured `Observability` instance.
    """
    return cls(debug=debug)