Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/infra/serializer.py: 99%
155 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-19 23:36 +0000
1# SPDX-License-Identifier: MIT
2# Copyright © 2025 Bijan Mousavi
4"""Provides concrete serialization services for JSON and YAML formats.
6This module defines concrete implementations of the `SerializerProtocol`. It
7offers different serializers optimized for performance and specific formats,
8gracefully handling optional dependencies.
10Key components include:
11 * `OrjsonSerializer`: A high-performance serializer that uses `orjson` for
12 JSON serialization if installed, falling back to the standard `json`
13 module. It uses `PyYAML` for YAML.
14 * `PyYAMLSerializer`: A serializer that exclusively handles the YAML format.
15 * `Redacted`: A special string subclass for wrapping sensitive data to prevent
16 it from being exposed in serialized output.
17 * `serializer_for`: A factory function that returns the most appropriate
18 serializer instance for a given format.
19"""
21from __future__ import annotations
23import abc
24import importlib.util as _importlib_util
25import json
26import sys
27from types import ModuleType
28import typing
29from typing import TYPE_CHECKING, Any, Final
31from injector import inject
33from bijux_cli.contracts import TelemetryProtocol
34from bijux_cli.core.enums import OutputFormat
35from bijux_cli.core.exceptions import BijuxError
37_orjson_spec = _importlib_util.find_spec("orjson")
38_yaml_spec = _importlib_util.find_spec("yaml")
40_orjson_mod: ModuleType | None
41try:
42 import orjson as _orjson_mod
43except ImportError:
44 _orjson_mod = None
45_ORJSON: Final[ModuleType | None] = _orjson_mod
47_yaml_mod: ModuleType | None
48try:
49 import yaml as _yaml_mod
50except ImportError:
51 _yaml_mod = None
52_YAML: Final[ModuleType | None] = _yaml_mod
54_HAS_ORJSON: Final[bool] = _ORJSON is not None
55_HAS_YAML: Final[bool] = _YAML is not None
58def yaml_dump(obj: Any, pretty: bool) -> str:
59 """Dumps an object to a YAML string using PyYAML.
61 Args:
62 obj (Any): The object to serialize.
63 pretty (bool): If True, formats the output in an indented block style.
65 Returns:
66 str: The serialized YAML string.
68 Raises:
69 BijuxError: If the `PyYAML` library is not installed.
70 """
71 if _yaml_mod is None:
72 raise BijuxError("PyYAML is required for YAML operations")
73 dumped = _yaml_mod.safe_dump(
74 obj,
75 sort_keys=False,
76 default_flow_style=not pretty,
77 indent=2 if pretty else None,
78 )
79 return dumped or ""
82class Redacted(str):
83 """A string subclass that redacts its value when printed or serialized.
85 This is used to wrap sensitive data, such as secrets or API keys, to
86 prevent them from being accidentally exposed in logs or console output.
87 """
89 def __new__(cls, value: str) -> Redacted:
90 """Creates a new `Redacted` string instance.
92 Args:
93 value (str): The original, sensitive value to be wrapped.
95 Returns:
96 Redacted: The new `Redacted` string instance.
97 """
98 return str.__new__(cls, value)
100 def __str__(self) -> str:
101 """Returns the redacted representation.
103 Returns:
104 str: A static string "***" to represent the redacted value.
105 """
106 return "***"
108 @staticmethod
109 def to_json() -> str:
110 """Provides a JSON-serializable representation for libraries like `orjson`.
112 Returns:
113 str: A static string "***" to represent the redacted value.
114 """
115 return "***"
118class _Base(abc.ABC):
119 """An abstract base class for all serializer implementations.
121 Attributes:
122 _telemetry (TelemetryProtocol | None): The telemetry service for events.
123 """
125 def __init__(self, telemetry: TelemetryProtocol | None) -> None:
126 """Initializes the base serializer.
128 Args:
129 telemetry (TelemetryProtocol | None): The telemetry service for
130 tracking serialization events.
131 """
132 self._telemetry = telemetry
134 def emit(
135 self,
136 payload: Any,
137 *,
138 fmt: OutputFormat = OutputFormat.JSON,
139 pretty: bool = False,
140 ) -> None:
141 """Serializes a payload and writes it to standard output.
143 Args:
144 payload (Any): The object to serialize and emit.
145 fmt (OutputFormat): The output format.
146 pretty (bool): If True, formats the output for human readability.
148 Returns:
149 None:
150 """
151 sys.stdout.write(self.dumps(payload, fmt=fmt, pretty=pretty))
152 if not sys.stdout.isatty():
153 sys.stdout.write("\n")
154 sys.stdout.flush()
156 @abc.abstractmethod
157 def dumps(
158 self,
159 obj: Any,
160 *,
161 fmt: OutputFormat,
162 pretty: bool = False,
163 ) -> str:
164 """Serializes an object to a string.
166 Args:
167 obj (Any): The object to serialize.
168 fmt (OutputFormat): The desired output format.
169 pretty (bool): If True, formats the output for human readability.
171 Returns:
172 str: The serialized string.
173 """
174 ...
176 @abc.abstractmethod
177 def dumps_bytes(
178 self,
179 obj: Any,
180 *,
181 fmt: OutputFormat,
182 pretty: bool = False,
183 ) -> bytes:
184 """Serializes an object to bytes.
186 Args:
187 obj (Any): The object to serialize.
188 fmt (OutputFormat): The desired output format.
189 pretty (bool): If True, formats the output for human readability.
191 Returns:
192 bytes: The serialized bytes.
193 """
194 ...
196 @abc.abstractmethod
197 def loads(
198 self,
199 data: str | bytes,
200 *,
201 fmt: OutputFormat,
202 pretty: bool = False,
203 ) -> Any:
204 """Deserializes data into a Python object.
206 Args:
207 data (str | bytes): The string or bytes to deserialize.
208 fmt (OutputFormat): The format of the input data.
209 pretty (bool): A hint that may affect parsing (often unused).
211 Returns:
212 Any: The deserialized object.
213 """
214 ...
216 def _event(self, name: str, **data: Any) -> None:
217 """Sends a telemetry event if the telemetry service is configured.
219 Args:
220 name (str): The name of the event.
221 **data (Any): Additional data for the event payload.
223 Returns:
224 None:
225 """
226 if self._telemetry is not None: 226 ↛ exitline 226 didn't return from function '_event' because the condition on line 226 was always true
227 self._telemetry.event(name, data)
229 @staticmethod
230 def _axerr(fmt: OutputFormat, action: str, exc: Exception) -> BijuxError:
231 """Creates a standardized `BijuxError` for serialization failures.
233 Args:
234 fmt (OutputFormat): The format that was being processed.
235 action (str): The action being performed (e.g., "serialize").
236 exc (Exception): The original exception that was caught.
238 Returns:
239 BijuxError: The wrapped error.
240 """
241 return BijuxError(f"Failed to {action} {fmt.value}: {exc}")
244class OrjsonSerializer(_Base):
245 """A serializer that uses `orjson` for JSON and `PyYAML` for YAML.
247 This implementation prioritizes performance by using `orjson` for JSON
248 operations if it is installed, gracefully falling back to the standard
249 library's `json` module if it is not.
250 """
252 @staticmethod
253 def _default(obj: Any) -> Any:
254 """Provides a default function for JSON serialization.
256 This handler allows custom types like `Redacted` to be serialized.
258 Args:
259 obj (Any): The object being serialized.
261 Returns:
262 Any: A serializable representation of the object.
264 Raises:
265 TypeError: If the object is not of a known custom type and is
266 not otherwise JSON serializable.
267 """
268 if isinstance(obj, Redacted):
269 return str(obj)
270 raise TypeError(f"{obj!r} is not JSON serialisable")
272 @staticmethod
273 def _yaml_dump(obj: Any, pretty: bool) -> str:
274 """Dumps an object to a YAML string.
276 Args:
277 obj (Any): The object to serialize.
278 pretty (bool): If True, formats the output in block style.
280 Returns:
281 str: The serialized YAML string.
283 Raises:
284 BijuxError: If the `PyYAML` library is not installed.
285 """
286 if not _HAS_YAML:
287 raise BijuxError("PyYAML is required for YAML operations")
288 assert _YAML is not None # noqa: S101 # nosec B101
289 dumped = _YAML.safe_dump(
290 obj,
291 sort_keys=False,
292 default_flow_style=not pretty,
293 indent=2 if pretty else None,
294 )
295 return dumped or ""
297 def _json_dump(self, obj: Any, pretty: bool) -> str | bytes:
298 """Dumps an object to a JSON string or bytes, preferring `orjson`.
300 Args:
301 obj (Any): The object to serialize.
302 pretty (bool): If True, indents the output.
304 Returns:
305 str | bytes: The serialized JSON data. Returns `bytes` if `orjson`
306 is used, otherwise returns `str`.
307 """
308 if _HAS_ORJSON:
309 assert _ORJSON is not None # noqa: S101 # nosec B101
310 opts = _ORJSON.OPT_INDENT_2 if pretty else 0
311 raw = _ORJSON.dumps(obj, option=opts, default=self._default)
312 return typing.cast(bytes, raw) # pyright: ignore[reportUnnecessaryCast]
313 return json.dumps(
314 obj,
315 indent=2 if pretty else None,
316 ensure_ascii=False,
317 default=self._default,
318 )
320 def dumps(
321 self,
322 obj: Any,
323 *,
324 fmt: OutputFormat = OutputFormat.JSON,
325 pretty: bool = False,
326 ) -> str:
327 """Serializes an object to a string.
329 Args:
330 obj (Any): The object to serialize.
331 fmt (OutputFormat): The desired output format.
332 pretty (bool): If True, formats the output for human readability.
334 Returns:
335 str: The serialized string.
337 Raises:
338 BijuxError: If the format is unsupported or serialization fails.
339 """
340 try:
341 if fmt is OutputFormat.JSON:
342 raw = self._json_dump(obj, pretty)
343 res = raw if isinstance(raw, str) else raw.decode()
344 elif fmt is OutputFormat.YAML:
345 res = self._yaml_dump(obj, pretty)
346 else:
347 raise BijuxError(f"Unsupported format: {fmt}")
348 self._event("serialize_dumps", format=fmt.value, pretty=pretty)
349 return res
350 except Exception as exc:
351 self._event("serialize_dumps_failed", format=fmt.value, error=str(exc))
352 raise self._axerr(fmt, "serialize", exc) from exc
354 def dumps_bytes(
355 self,
356 obj: Any,
357 *,
358 fmt: OutputFormat = OutputFormat.JSON,
359 pretty: bool = False,
360 ) -> bytes:
361 """Serializes an object to bytes.
363 Args:
364 obj (Any): The object to serialize.
365 fmt (OutputFormat): The desired output format.
366 pretty (bool): If True, formats the output for human readability.
368 Returns:
369 bytes: The serialized bytes.
371 Raises:
372 BijuxError: If the format is unsupported or serialization fails.
373 """
374 try:
375 if fmt is OutputFormat.JSON:
376 raw = self._json_dump(obj, pretty)
377 res = raw if isinstance(raw, bytes) else raw.encode()
378 elif fmt is OutputFormat.YAML:
379 res = self.dumps(obj, fmt=fmt, pretty=pretty).encode()
380 else:
381 raise BijuxError(f"Unsupported format: {fmt}")
382 self._event("serialize_dumps_bytes", format=fmt.value, pretty=pretty)
383 return res
384 except Exception as exc:
385 self._event(
386 "serialize_dumps_bytes_failed",
387 format=fmt.value,
388 error=str(exc),
389 )
390 raise self._axerr(fmt, "serialize", exc) from exc
392 def loads(
393 self,
394 data: str | bytes,
395 *,
396 fmt: OutputFormat = OutputFormat.JSON,
397 pretty: bool = False,
398 ) -> Any:
399 """Deserializes data into a Python object.
401 Args:
402 data (str | bytes): The string or bytes to deserialize.
403 fmt (OutputFormat): The format of the input data.
404 pretty (bool): A hint that may affect parsing (often unused).
406 Returns:
407 Any: The deserialized object.
409 Raises:
410 BijuxError: If the format is unsupported or deserialization fails.
411 """
412 try:
413 if fmt is OutputFormat.JSON:
414 if _HAS_ORJSON:
415 assert _ORJSON is not None # noqa: S101 # nosec B101
416 res = _ORJSON.loads(data)
417 else:
418 res = json.loads(data)
419 elif fmt is OutputFormat.YAML:
420 if not _HAS_YAML:
421 raise BijuxError("PyYAML is required for YAML operations")
422 assert _YAML is not None # noqa: S101 # nosec B101
423 txt = data if isinstance(data, str) else data.decode()
424 res = _YAML.safe_load(txt) or {}
425 else:
426 raise BijuxError(f"Unsupported format: {fmt}")
427 self._event("serialize_loads", format=fmt.value)
428 return res
429 except Exception as exc:
430 self._event("serialize_loads_failed", format=fmt.value, error=str(exc))
431 raise self._axerr(fmt, "deserialize", exc) from exc
434class PyYAMLSerializer(_Base):
435 """A serializer that exclusively uses the `PyYAML` library for YAML.
437 Attributes:
438 _patched (bool): A class-level flag to ensure that custom YAML
439 representers are only registered once.
440 """
442 _patched = False
444 @inject
445 def __init__(self, telemetry: TelemetryProtocol | None) -> None:
446 """Initializes the `PyYAMLSerializer`.
448 This also registers a custom YAML representer for the `Redacted` type
449 on first instantiation.
451 Args:
452 telemetry (TelemetryProtocol | None): The telemetry service.
454 Raises:
455 BijuxError: If the `PyYAML` library is not installed.
456 """
457 if not _HAS_YAML:
458 raise BijuxError("PyYAML is not installed")
459 super().__init__(telemetry)
460 if not PyYAMLSerializer._patched:
461 assert _YAML is not None # noqa: S101 # nosec B101
462 _YAML.add_representer(
463 Redacted,
464 lambda dumper, data: dumper.represent_scalar(
465 "tag:yaml.org,2002:str", str(data)
466 ),
467 Dumper=_YAML.SafeDumper,
468 )
469 PyYAMLSerializer._patched = True
471 def dumps(
472 self,
473 obj: Any,
474 *,
475 fmt: OutputFormat = OutputFormat.YAML,
476 pretty: bool = False,
477 ) -> str:
478 """Serializes an object to a YAML string.
480 Args:
481 obj (Any): The object to serialize.
482 fmt (OutputFormat): The output format. Must be `OutputFormat.YAML`.
483 pretty (bool): If True, formats the output in block style.
485 Returns:
486 str: The serialized YAML string.
488 Raises:
489 BijuxError: If the format is not `OutputFormat.YAML`.
490 """
491 if fmt is not OutputFormat.YAML:
492 raise BijuxError("PyYAMLSerializer only supports YAML")
493 return yaml_dump(obj, pretty)
495 def dumps_bytes(
496 self,
497 obj: Any,
498 *,
499 fmt: OutputFormat = OutputFormat.YAML,
500 pretty: bool = False,
501 ) -> bytes:
502 """Serializes an object to YAML bytes.
504 Args:
505 obj (Any): The object to serialize.
506 fmt (OutputFormat): The output format. Must be `OutputFormat.YAML`.
507 pretty (bool): If True, formats the output in block style.
509 Returns:
510 bytes: The serialized YAML bytes.
511 """
512 return self.dumps(obj, fmt=fmt, pretty=pretty).encode()
514 def loads(
515 self,
516 data: str | bytes,
517 *,
518 fmt: OutputFormat = OutputFormat.YAML,
519 pretty: bool = False,
520 ) -> Any:
521 """Deserializes YAML data into a Python object.
523 Args:
524 data (str | bytes): The string or bytes to deserialize.
525 fmt (OutputFormat): The format of the input. Must be `OutputFormat.YAML`.
526 pretty (bool): A hint that may affect parsing (unused).
528 Returns:
529 Any: The deserialized object.
531 Raises:
532 BijuxError: If the format is not `OutputFormat.YAML`.
533 """
534 if fmt is not OutputFormat.YAML:
535 raise BijuxError("PyYAMLSerializer only supports YAML")
536 txt = data if isinstance(data, str) else data.decode()
537 assert _YAML is not None # noqa: S101 # nosec B101
538 return _YAML.safe_load(txt) or {}
541if TYPE_CHECKING:
542 from bijux_cli.contracts import SerializerProtocol
545def serializer_for(
546 fmt: OutputFormat | str,
547 telemetry: TelemetryProtocol,
548) -> SerializerProtocol[Any]:
549 """A factory function that returns a serializer for the given format.
551 Args:
552 fmt (OutputFormat | str): The desired output format.
553 telemetry (TelemetryProtocol): The telemetry service to inject into
554 the serializer.
556 Returns:
557 SerializerProtocol[Any]: A configured serializer instance appropriate
558 for the specified format.
559 """
560 format_enum = fmt if isinstance(fmt, OutputFormat) else OutputFormat(fmt.upper())
562 if format_enum is OutputFormat.JSON:
563 return OrjsonSerializer(telemetry)
564 else:
565 return PyYAMLSerializer(telemetry)
568__all__ = [
569 "Redacted",
570 "OrjsonSerializer",
571 "PyYAMLSerializer",
572 "serializer_for",
573]