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

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Serialization adapters for JSON and YAML formats.""" 

5 

6from __future__ import annotations 

7 

8import importlib.util as _importlib_util 

9import json 

10from types import ModuleType 

11from typing import Any, Final, cast 

12 

13from bijux_cli.core.enums import OutputFormat 

14 

15_orjson_spec = _importlib_util.find_spec("orjson") 

16_yaml_spec = _importlib_util.find_spec("yaml") 

17 

18_orjson_mod: ModuleType | None 

19try: 

20 import orjson as _orjson_mod 

21except ImportError: 

22 _orjson_mod = None 

23_ORJSON: Final[ModuleType | None] = _orjson_mod 

24 

25_yaml_mod: ModuleType | None 

26try: 

27 import yaml as _yaml_mod 

28except ImportError: 

29 _yaml_mod = None 

30_YAML: Final[ModuleType | None] = _yaml_mod 

31 

32 

33class SerializationError(RuntimeError): 

34 """Raised when serialization or deserialization fails.""" 

35 

36 

37def _yaml_dump(obj: Any, pretty: bool) -> str: 

38 """Serialize an object to YAML.""" 

39 if _YAML is None: 

40 raise SerializationError("PyYAML is required for YAML operations") 

41 dumped = _YAML.safe_dump( 

42 obj, 

43 sort_keys=False, 

44 default_flow_style=not pretty, 

45 indent=2 if pretty else None, 

46 ) 

47 return dumped or "" 

48 

49 

50class OrjsonSerializer: 

51 """Serializer that handles JSON (and YAML via PyYAML).""" 

52 

53 def __init__(self, telemetry: Any | None) -> None: 

54 """Initialize with telemetry.""" 

55 self._telemetry = telemetry 

56 

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

58 """Serialize an object to JSON or YAML.""" 

59 if fmt is OutputFormat.JSON: 

60 try: 

61 if _ORJSON is not None: 

62 option = _ORJSON.OPT_INDENT_2 if pretty else 0 

63 return cast( 

64 str, 

65 _ORJSON.dumps(obj, option=option).decode("utf-8"), 

66 ) 

67 return json.dumps(obj, indent=2 if pretty else None) 

68 except Exception as exc: 

69 raise SerializationError(f"Failed to serialize json: {exc}") from exc 

70 if fmt is OutputFormat.YAML: 

71 return _yaml_dump(obj, pretty) 

72 raise SerializationError(f"Unsupported format: {fmt}") 

73 

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

75 """Serialize an object to bytes.""" 

76 return self.dumps(obj, fmt=fmt, pretty=pretty).encode("utf-8") 

77 

78 def loads( 

79 self, 

80 data: str | bytes, 

81 *, 

82 fmt: OutputFormat, 

83 pretty: bool, 

84 ) -> Any: 

85 """Deserialize JSON or YAML data.""" 

86 if fmt is OutputFormat.JSON: 

87 try: 

88 return json.loads(data) 

89 except Exception as exc: 

90 raise SerializationError(f"Failed to deserialize json: {exc}") from exc 

91 if fmt is OutputFormat.YAML: 

92 if _YAML is None: 

93 raise SerializationError("PyYAML is required for YAML operations") 

94 return _YAML.safe_load(data) 

95 raise SerializationError(f"Unsupported format: {fmt}") 

96 

97 

98class PyYAMLSerializer: 

99 """Serializer restricted to YAML format.""" 

100 

101 def __init__(self, telemetry: Any | None) -> None: 

102 """Initialize with telemetry.""" 

103 if _YAML is None: 

104 raise SerializationError("PyYAML is not installed") 

105 self._telemetry = telemetry 

106 

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

108 """Serialize an object to YAML.""" 

109 if fmt is not OutputFormat.YAML: 

110 raise SerializationError("PyYAMLSerializer only supports YAML") 

111 return _yaml_dump(obj, pretty) 

112 

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

114 """Serialize an object to bytes.""" 

115 return self.dumps(obj, fmt=fmt, pretty=pretty).encode("utf-8") 

116 

117 def loads( 

118 self, 

119 data: str | bytes, 

120 *, 

121 fmt: OutputFormat, 

122 pretty: bool, 

123 ) -> Any: 

124 """Deserialize YAML data.""" 

125 if fmt is not OutputFormat.YAML: 

126 raise SerializationError("PyYAMLSerializer only supports YAML") 

127 return _YAML.safe_load(data) if _YAML is not None else None 

128 

129 

130def serializer_for( 

131 fmt: OutputFormat, telemetry: Any | None 

132) -> OrjsonSerializer | PyYAMLSerializer: 

133 """Return the best serializer for the requested format.""" 

134 if fmt is OutputFormat.JSON: 

135 return OrjsonSerializer(telemetry) 

136 if fmt is OutputFormat.YAML: 

137 return PyYAMLSerializer(telemetry) 

138 raise SerializationError(f"Unsupported format: {fmt}") 

139 

140 

141__all__ = [ 

142 "SerializationError", 

143 "OrjsonSerializer", 

144 "PyYAMLSerializer", 

145 "serializer_for", 

146]