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

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Infrastructure protocol contracts for Bijux CLI.""" 

5 

6from __future__ import annotations 

7 

8from typing import Any, Protocol, runtime_checkable 

9 

10from bijux_cli.core.enums import LogLevel, OutputFormat 

11 

12 

13@runtime_checkable 

14class Serializer(Protocol): 

15 """Serializer adapter for structured output.""" 

16 

17 def dumps(self, obj: Any, *, fmt: OutputFormat, pretty: bool) -> str: 

18 """Serialize data to a string.""" 

19 ... 

20 

21 def dumps_bytes(self, obj: Any, *, fmt: OutputFormat, pretty: bool) -> bytes: 

22 """Serialize data to bytes.""" 

23 ... 

24 

25 def loads(self, data: str | bytes, *, fmt: OutputFormat, pretty: bool) -> Any: 

26 """Deserialize data into a value.""" 

27 ... 

28 

29 

30@runtime_checkable 

31class RetryPolicy(Protocol): 

32 """Retry policy for transient failures.""" 

33 

34 def run(self, fn: Any, *args: Any, **kwargs: Any) -> Any: 

35 """Execute a callable with retry behavior.""" 

36 ... 

37 

38 def reset(self) -> None: 

39 """Reset any internal retry state.""" 

40 ... 

41 

42 

43@runtime_checkable 

44class Emitter(Protocol): 

45 """Emitter for structured output.""" 

46 

47 def emit( 

48 self, 

49 payload: Any, 

50 *, 

51 fmt: OutputFormat, 

52 pretty: bool, 

53 level: LogLevel, 

54 message: str, 

55 output: str | None, 

56 emit_output: bool = True, 

57 emit_diagnostics: bool = False, 

58 **context: Any, 

59 ) -> None: 

60 """Serialize and emit a structured payload.""" 

61 ... 

62 

63 def flush(self) -> None: 

64 """Flush any buffered output.""" 

65 ... 

66 

67 

68@runtime_checkable 

69class ProcessRunner(Protocol): 

70 """Runner for isolated command execution.""" 

71 

72 def run(self, cmd: list[str], *, executor: str) -> tuple[int, bytes, bytes]: 

73 """Run a command with an executor.""" 

74 ... 

75 

76 def shutdown(self) -> None: 

77 """Shut down the runner.""" 

78 ... 

79 

80 def get_status(self) -> dict[str, Any]: 

81 """Return runner status info.""" 

82 ... 

83 

84 

85__all__ = [ 

86 "Emitter", 

87 "ProcessRunner", 

88 "RetryPolicy", 

89 "Serializer", 

90]