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 

17 

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

19 

20 

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

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

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

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

25 try: 

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

27 except OSError: 

28 return "" 

29 

30 

31def _parse_every_manifest( 

32 root_path: pathlib.Path, 

33 configured_manifest_dir: str | pathlib.Path | None, 

34 /, 

35) -> _ParsedManifests: 

36 return tuple( 

37 ( 

38 one_manifest_functions, 

39 one_manifest_functions.parse_manifests( 

40 one_manifest_functions.read_source(root_path, configured_manifest_dir), 

41 ), 

42 ) 

43 for one_manifest_functions in mapping.MAPPING_OF_MANIFEST_PARSERS_AND_RENDERERS.values() 

44 ) 

45 

46 

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

48 return tuple( 

49 one_annotation 

50 for one_manifest_functions, one_parsed_manifest in all_parsed_manifests 

51 for one_annotation in one_manifest_functions.render_node_annotations(one_parsed_manifest) 

52 ) 

53 

54 

55def _render_manifest_edges( 

56 service_node: diagram_model.DiagramNode, 

57 all_parsed_manifests: _ParsedManifests, 

58 /, 

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

60 return tuple( 

61 one_edge 

62 for one_manifest_functions, one_parsed_manifest in all_parsed_manifests 

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

64 ) 

65 

66 

67@typing.final 

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

69class ArchitectureParserAndRenderer: 

70 local_settings: SettingsForArchdocs 

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

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

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

74 

75 def render_architecture_diagram(self) -> str: 

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

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

78 if self._rendered_diagram_cache: 

79 return self._rendered_diagram_cache[0] 

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

81 self._rendered_diagram_cache.append(rendered_diagram) 

82 return rendered_diagram 

83 

84 def _render_every_diagram_line(self) -> str: 

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

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

87 service_node: typing.Final = diagram_model.build_service_node( 

88 self.local_settings.service_name, 

89 _render_manifest_annotations(all_parsed_manifests), 

90 ) 

91 all_edges: typing.Final = ( 

92 *_render_manifest_edges(service_node, all_parsed_manifests), 

93 *self._render_source_edges(service_node, root_path), 

94 ) 

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

96 

97 def _render_source_edges( 

98 self, 

99 service_node: diagram_model.DiagramNode, 

100 root_path: pathlib.Path, 

101 /, 

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

103 py_files: typing.Final = sorted( 

104 one_source_file 

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

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

107 ) 

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

109 all_file_edges: typing.Final = executor.map( 

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

111 py_files, 

112 ) 

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

114 

115 def _render_one_file_edges( 

116 self, 

117 service_node: diagram_model.DiagramNode, 

118 one_source_file: pathlib.Path, 

119 /, 

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

121 raw_file_source: typing.Final = _read_source_text(one_source_file) 

122 return tuple( 

123 one_edge 

124 for one_feature_functions in mapping.MAPPING_OF_PARSERS_AND_RENDERERS.values() 

125 for one_edge in one_feature_functions.render_edges( 

126 service_node, 

127 one_feature_functions.parse_source(raw_file_source), 

128 ) 

129 )