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

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Provides the concrete implementation of the API specification writing service. 

5 

6This module defines the `Docs` class, which implements the `DocsProtocol`. 

7It is responsible for serializing API specification data into formats like 

8JSON or YAML and writing the resulting documents to the filesystem. It 

9integrates with observability and telemetry services to log its activities. 

10""" 

11 

12from __future__ import annotations 

13 

14import os 

15from pathlib import Path 

16from typing import Any 

17from weakref import WeakKeyDictionary 

18 

19from injector import inject 

20 

21from bijux_cli.contracts import DocsProtocol, ObservabilityProtocol, TelemetryProtocol 

22from bijux_cli.core.enums import OutputFormat 

23from bijux_cli.core.exceptions import ServiceError 

24from bijux_cli.infra.serializer import serializer_for 

25 

26 

27class Docs(DocsProtocol): 

28 """A service for writing API specification documents to disk. 

29 

30 This class implements the `DocsProtocol` to handle the serialization and 

31 writing of specifications (e.g., OpenAPI, JSON Schema) to files. It 

32 maintains a cache of serializer instances for performance. 

33 

34 Attributes: 

35 _serializers (WeakKeyDictionary): A cache of serializer instances, keyed 

36 by the telemetry service instance. 

37 _observability (ObservabilityProtocol): The logging service. 

38 _telemetry (TelemetryProtocol): The telemetry service for event tracking. 

39 _root (Path): The root directory where documents will be written. 

40 """ 

41 

42 _serializers: WeakKeyDictionary[TelemetryProtocol, dict[OutputFormat, Any]] = ( 

43 WeakKeyDictionary() 

44 ) 

45 

46 @inject 

47 def __init__( 

48 self, 

49 observability: ObservabilityProtocol, 

50 telemetry: TelemetryProtocol, 

51 root: str | Path | None = None, 

52 ) -> None: 

53 """Initializes the `Docs` service. 

54 

55 Args: 

56 observability (ObservabilityProtocol): The service for logging. 

57 telemetry (TelemetryProtocol): The service for event tracking. 

58 root (str | Path | None): The root directory for writing documents. 

59 It defaults to the `BIJUXCLI_DOCS_DIR` environment variable, 

60 or "docs" if not set. 

61 """ 

62 self._observability = observability 

63 self._telemetry = telemetry 

64 env_root = os.getenv("BIJUXCLI_DOCS_DIR") 

65 root_dir = env_root if env_root else (root or "docs") 

66 self._root = Path(root_dir) 

67 self._root.mkdir(exist_ok=True, parents=True) 

68 if telemetry not in self._serializers: 

69 self._serializers[telemetry] = {} 

70 

71 def render(self, spec: dict[str, Any], *, fmt: OutputFormat) -> str: 

72 """Renders a specification dictionary to a string in the given format. 

73 

74 Args: 

75 spec (dict[str, Any]): The specification dictionary to serialize. 

76 fmt (OutputFormat): The desired output format (e.g., JSON, YAML). 

77 

78 Returns: 

79 str: The serialized specification as a string. 

80 

81 Raises: 

82 TypeError: If the underlying serializer returns a non-string result. 

83 """ 

84 if self._telemetry not in self._serializers: 

85 self._serializers[self._telemetry] = {} 

86 if fmt not in self._serializers[self._telemetry]: 

87 self._serializers[self._telemetry][fmt] = serializer_for( 

88 fmt, self._telemetry 

89 ) 

90 result = self._serializers[self._telemetry][fmt].dumps( 

91 spec, fmt=fmt, pretty=False 

92 ) 

93 if not isinstance(result, str): 

94 raise TypeError( 

95 f"Expected str from serializer.dumps, got {type(result).__name__}" 

96 ) 

97 return result 

98 

99 def write( 

100 self, 

101 spec: dict[str, Any], 

102 *, 

103 fmt: OutputFormat = OutputFormat.JSON, 

104 name: str = "spec", 

105 ) -> str: 

106 """Writes a specification to a file and returns the path as a string. 

107 

108 This is a convenience wrapper around `write_sync`. 

109 

110 Args: 

111 spec (dict[str, Any]): The specification dictionary to write. 

112 fmt (OutputFormat): The output format. Defaults to `OutputFormat.JSON`. 

113 name (str): The base name for the output file. Defaults to 'spec'. 

114 

115 Returns: 

116 str: The absolute path to the written file. 

117 """ 

118 path = self.write_sync(spec, fmt, name) 

119 return str(path) 

120 

121 def write_sync( 

122 self, spec: dict[str, Any], fmt: OutputFormat, name: str | Path 

123 ) -> Path: 

124 """Writes the specification to a file synchronously. 

125 

126 This method handles path resolution, serializes the `spec` dictionary, 

127 and writes the content to the final destination file. 

128 

129 Args: 

130 spec (dict[str, Any]): The specification dictionary to write. 

131 fmt (OutputFormat): The desired output format. 

132 name (str | Path): The path or base name for the output file. 

133 

134 Returns: 

135 Path: The `Path` object pointing to the written file. 

136 

137 Raises: 

138 ServiceError: If writing to the file fails due to an `OSError`. 

139 """ 

140 final_path = None 

141 try: 

142 final_path = Path(name).expanduser().resolve() 

143 if final_path.is_dir(): 

144 final_path = final_path / f"spec.{fmt.value}" 

145 final_path.parent.mkdir(parents=True, exist_ok=True) 

146 content = self.render(spec, fmt=fmt) 

147 final_path.write_text(content, encoding="utf-8") 

148 self._observability.log("info", f"Wrote docs to {final_path}") 

149 self._telemetry.event( 

150 "docs_written", {"path": str(final_path), "format": fmt.value} 

151 ) 

152 return final_path 

153 except OSError as exc: 

154 self._telemetry.event( 

155 "docs_write_failed", 

156 { 

157 "path": ( 

158 str(final_path) if final_path is not None else "<unresolved>" 

159 ), 

160 "error": str(exc), 

161 }, 

162 ) 

163 raise ServiceError(f"Unable to write spec: {exc}", http_status=403) from exc 

164 

165 def close(self) -> None: 

166 """Closes the service. This is a no-op for this implementation.""" 

167 return 

168 

169 

170__all__ = ["Docs"]