Coverage for archdocs/main.py: 100%

1import dataclasses 

2import functools 

3import pathlib 

4import typing 

5from concurrent import futures 

6 

7from archdocs import diagram_model, mapping, mermaid_syntax, settings 

8 

9 

10@typing.final 

11@dataclasses.dataclass(frozen=True, slots=True, kw_only=True) 

12class SettingsForArchdocs: 

13 root_dir: str | pathlib.Path 

14 service_name: str 

15 kubernetes_dir: str | pathlib.Path | None = None 

16 # Both assets come from public CDNs by default; a network that cannot reach them serves its 

17 # own copy of mermaid and drops the font, which falls back to the system sans-serif. 

18 mermaid_script_url: str = settings.DEFAULT_MERMAID_SCRIPT_URL 

19 font_stylesheet_url: str | None = settings.DEFAULT_FONT_STYLESHEET_URL 

20 

21 

22type _ParsedManifests = tuple[tuple[mapping.ManifestFeatureFunctions[typing.Any], typing.Any], ...] 

23 

24 

25def _read_source_text(one_source_file: pathlib.Path, /) -> str: 

26 # The scanned tree is somebody's whole project, not a curated corpus: a source in a legacy 

27 # encoding, a symlink into nowhere or a file the process may not open costs that one file and 

28 # nothing else. Without this the page answers 500 instead of drawing the rest of the service. 

29 try: 

30 return one_source_file.read_text(errors="ignore") 

31 except OSError: 

32 return "" 

33 

34 

35def _parse_every_manifest( 

36 root_path: pathlib.Path, 

37 configured_manifest_dir: str | pathlib.Path | None, 

38 /, 

39) -> _ParsedManifests: 

40 return tuple( 

41 ( 

42 one_manifest_functions, 

43 one_manifest_functions.parse_manifests( 

44 one_manifest_functions.read_source(root_path, configured_manifest_dir), 

45 ), 

46 ) 

47 for one_manifest_functions in mapping.MAPPING_OF_MANIFEST_PARSERS_AND_RENDERERS.values() 

48 ) 

49 

50 

51def _render_manifest_annotations(all_parsed_manifests: _ParsedManifests, /) -> tuple[str, ...]: 

52 return tuple( 

53 one_annotation 

54 for one_manifest_functions, one_parsed_manifest in all_parsed_manifests 

55 for one_annotation in one_manifest_functions.render_node_annotations(one_parsed_manifest) 

56 ) 

57 

58 

59def _render_manifest_edges( 

60 service_node: diagram_model.DiagramNode, 

61 all_parsed_manifests: _ParsedManifests, 

62 /, 

63) -> tuple[diagram_model.DiagramEdge, ...]: 

64 return tuple( 

65 one_edge 

66 for one_manifest_functions, one_parsed_manifest in all_parsed_manifests 

67 for one_edge in one_manifest_functions.render_edges(service_node, one_parsed_manifest) 

68 ) 

69 

70 

71@typing.final 

72@dataclasses.dataclass(frozen=True, slots=True, kw_only=True) 

73class ArchitectureParserAndRenderer: 

74 local_settings: SettingsForArchdocs 

75 # The cache is state, not identity: leaving it out of `__eq__` keeps the frozen dataclass 

76 # hashable, which a list field would otherwise take away. 

77 _rendered_diagram_cache: list[str] = dataclasses.field(default_factory=list, compare=False) 

78 

79 def render_architecture_diagram(self) -> str: 

80 # Not `functools.cache`: on a method it keeps the instance alive for the process 

81 # lifetime, which is exactly what ruff B019 forbids. 

82 if self._rendered_diagram_cache: 

83 return self._rendered_diagram_cache[0] 

84 rendered_diagram: typing.Final = self._render_every_diagram_line() 

85 self._rendered_diagram_cache.append(rendered_diagram) 

86 return rendered_diagram 

87 

88 def _render_every_diagram_line(self) -> str: 

89 root_path: typing.Final = pathlib.Path(self.local_settings.root_dir).resolve() 

90 all_parsed_manifests: typing.Final = _parse_every_manifest(root_path, self.local_settings.kubernetes_dir) 

91 service_node: typing.Final = diagram_model.build_service_node( 

92 self.local_settings.service_name, 

93 _render_manifest_annotations(all_parsed_manifests), 

94 ) 

95 all_edges: typing.Final = ( 

96 *_render_manifest_edges(service_node, all_parsed_manifests), 

97 *self._render_source_edges(service_node, root_path), 

98 ) 

99 return mermaid_syntax.MermaidDiagram(service_node=service_node, all_edges=all_edges).render_every_line() 

100 

101 def _render_source_edges( 

102 self, 

103 service_node: diagram_model.DiagramNode, 

104 root_path: pathlib.Path, 

105 /, 

106 ) -> tuple[diagram_model.DiagramEdge, ...]: 

107 py_files: typing.Final = sorted( 

108 one_source_file 

109 for one_source_file in root_path.rglob(settings.FILES_SEARCH_PATTERN) 

110 if settings.SKIPPED_DIR_NAMES.isdisjoint(one_source_file.relative_to(root_path).parts) 

111 ) 

112 with futures.ThreadPoolExecutor(max_workers=settings.MAX_WORKERS) as executor: 

113 all_file_edges: typing.Final = executor.map( 

114 functools.partial(self._render_one_file_edges, service_node), 

115 py_files, 

116 ) 

117 return tuple(one_edge for one_file_edges in all_file_edges for one_edge in one_file_edges) 

118 

119 def _render_one_file_edges( 

120 self, 

121 service_node: diagram_model.DiagramNode, 

122 one_source_file: pathlib.Path, 

123 /, 

124 ) -> tuple[diagram_model.DiagramEdge, ...]: 

125 raw_file_source: typing.Final = _read_source_text(one_source_file) 

126 return tuple( 

127 one_edge 

128 for one_feature_functions in mapping.MAPPING_OF_PARSERS_AND_RENDERERS.values() 

129 for one_edge in one_feature_functions.render_edges( 

130 service_node, 

131 one_feature_functions.parse_source(raw_file_source), 

132 ) 

133 )