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

1# SPDX-License-Identifier: Apache-2.0 

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 

14from collections.abc import Mapping 

15import os 

16from pathlib import Path 

17from typing import Any 

18 

19from injector import inject 

20 

21from bijux_cli.core.enums import OutputFormat 

22from bijux_cli.infra.contracts import Serializer 

23from bijux_cli.services.contracts import ObservabilityProtocol, TelemetryProtocol 

24from bijux_cli.services.diagnostics.contracts import DocsProtocol 

25from bijux_cli.services.errors import ServiceError 

26 

27 

28class Docs(DocsProtocol): 

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

30 

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

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

33 

34 Attributes: 

35 _observability (ObservabilityProtocol): The logging service. 

36 _serializer (Serializer): The serializer adapter for output. 

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

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

39 """ 

40 

41 @inject 

42 def __init__( 

43 self, 

44 observability: ObservabilityProtocol, 

45 serializer: Serializer, 

46 telemetry: TelemetryProtocol, 

47 root: str | Path | None = None, 

48 ) -> None: 

49 """Initializes the `Docs` service. 

50 

51 Args: 

52 observability (ObservabilityProtocol): The service for logging. 

53 serializer (Serializer): The serializer adapter for output. 

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

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

56 It defaults to the `BIJUXCLI_DOCS_DIR` environment variable, 

57 or "docs" if not set. 

58 """ 

59 self._observability = observability 

60 self._serializer = serializer 

61 self._telemetry = telemetry 

62 env_root = os.getenv("BIJUXCLI_DOCS_DIR") 

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

64 self._root = Path(root_dir) 

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

66 

67 def render( 

68 self, spec: Mapping[str, Any], *, fmt: OutputFormat, pretty: bool = False 

69 ) -> str: 

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

71 

72 Args: 

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

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

75 

76 Returns: 

77 str: The serialized specification as a string. 

78 

79 Raises: 

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

81 """ 

82 result = self._serializer.dumps(spec, fmt=fmt, pretty=pretty) 

83 if not isinstance(result, str): 

84 raise TypeError( 

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

86 ) 

87 return result 

88 

89 def write( 

90 self, 

91 spec: Mapping[str, Any], 

92 *, 

93 fmt: OutputFormat = OutputFormat.JSON, 

94 name: str = "spec", 

95 pretty: bool = False, 

96 ) -> str: 

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

98 

99 This is a convenience wrapper around `write_sync`. 

100 

101 Args: 

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

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

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

105 

106 Returns: 

107 str: The absolute path to the written file. 

108 """ 

109 path = self.write_sync(spec, fmt, name, pretty) 

110 return str(path) 

111 

112 def write_sync( 

113 self, 

114 spec: Mapping[str, Any], 

115 fmt: OutputFormat, 

116 name: str | Path, 

117 pretty: bool = False, 

118 ) -> Path: 

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

120 

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

122 and writes the content to the final destination file. 

123 

124 Args: 

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

126 fmt (OutputFormat): The desired output format. 

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

128 

129 Returns: 

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

131 

132 Raises: 

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

134 """ 

135 final_path = None 

136 try: 

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

138 if final_path.is_dir(): 

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

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

141 content = self.render(spec, fmt=fmt, pretty=pretty) 

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

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

144 self._telemetry.event( 

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

146 ) 

147 return final_path 

148 except OSError as exc: 

149 self._telemetry.event( 

150 "docs_write_failed", 

151 { 

152 "path": ( 

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

154 ), 

155 "error": str(exc), 

156 }, 

157 ) 

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

159 

160 def close(self) -> None: 

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

162 return 

163 

164 

165__all__ = ["Docs"]