Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / services / contracts.py: 100%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Service-level protocol contracts.""" 

5 

6from __future__ import annotations 

7 

8from collections.abc import Coroutine 

9from typing import Any, Protocol, Self, runtime_checkable 

10 

11from structlog.typing import FilteringBoundLogger 

12 

13from bijux_cli.core.enums import LogLevel 

14 

15 

16@runtime_checkable 

17class TelemetryProtocol(Protocol): 

18 """Fire-and-forget telemetry sink.""" 

19 

20 def event( 

21 self, name: str, payload: dict[str, Any] 

22 ) -> None | Coroutine[Any, Any, None]: 

23 """Record a telemetry event.""" 

24 ... 

25 

26 def flush(self) -> None: 

27 """Flush buffered events.""" 

28 ... 

29 

30 def enable(self) -> None: 

31 """Enable telemetry output.""" 

32 ... 

33 

34 

35@runtime_checkable 

36class ObservabilityProtocol(Protocol): 

37 """Structured logging facade.""" 

38 

39 @classmethod 

40 def setup(cls, *, log_level: LogLevel, telemetry: TelemetryProtocol) -> Self: 

41 """Configure and return a logging instance.""" 

42 ... 

43 

44 def get_logger(self) -> FilteringBoundLogger | None: 

45 """Return the bound logger instance.""" 

46 ... 

47 

48 def bind(self, **_kv: Any) -> Self: 

49 """Bind context keys for future log entries.""" 

50 ... 

51 

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

53 """Emit a structured log entry.""" 

54 ... 

55 

56 def close(self) -> None: 

57 """Close the logger and release resources.""" 

58 ... 

59 

60 def set_telemetry(self, telemetry: TelemetryProtocol) -> Self: 

61 """Attach a telemetry sink.""" 

62 ... 

63 

64 

65__all__ = [ 

66 "ObservabilityProtocol", 

67 "TelemetryProtocol", 

68]