Coverage for archdocs/main.py: 100%

1import dataclasses 

2import functools 

3import pathlib 

4import typing 

5from concurrent import futures 

6 

7from archdocs import diagram_model, mermaid_syntax, settings 

8from archdocs.mapping import ( 

9 MAPPING_OF_MANIFEST_PARSERS_AND_RENDERERS, 

10 MAPPING_OF_PARSERS_AND_RENDERERS, 

11 ManifestFeatureFunctions, 

12) 

13 

14 

15"""TODO: 

16 

17parsers from settings.py typical configuration 

18parsers from docker-compose.yml? 

19""" 

20 

21 

22@typing.final 

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

24class SettingsForArchdocs: 

25 root_dir: str | pathlib.Path 

26 service_name: str 

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

28 

29 

30type _ParsedManifests = tuple[tuple[ManifestFeatureFunctions, typing.Any], ...] 

31 

32 

33def _parse_every_manifest( 

34 root_path: pathlib.Path, 

35 configured_manifest_dir: str | pathlib.Path | None, 

36 /, 

37) -> _ParsedManifests: 

38 return tuple( 

39 ( 

40 one_manifest_functions, 

41 one_manifest_functions.parse_manifests( 

42 one_manifest_functions.read_source(root_path, configured_manifest_dir), 

43 ), 

44 ) 

45 for one_manifest_functions in MAPPING_OF_MANIFEST_PARSERS_AND_RENDERERS.values() 

46 ) 

47 

48 

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

50 return tuple( 

51 one_annotation 

52 for one_manifest_functions, one_parsed_manifest in all_parsed_manifests 

53 for one_annotation in one_manifest_functions.render_node_annotations(one_parsed_manifest) 

54 ) 

55 

56 

57def _render_manifest_edges( 

58 service_node: diagram_model.DiagramNode, 

59 all_parsed_manifests: _ParsedManifests, 

60 /, 

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

62 return tuple( 

63 one_edge 

64 for one_manifest_functions, one_parsed_manifest in all_parsed_manifests 

65 for one_edge in one_manifest_functions.render_diagram(service_node, one_parsed_manifest) 

66 ) 

67 

68 

69@typing.final 

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

71class ArchitectureParserAndRenderer: 

72 local_settings: SettingsForArchdocs 

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

74 

75 def render_architecture_diagram(self) -> str: 

76 # "why you doesnt use functools.cache lol" 

77 # https://docs.astral.sh/ruff/rules/cached-instance-method/#cached-instance-method-b019 

78 if self._rendered_diagram_cache: 

79 return self._rendered_diagram_cache[0] 

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

81 self._rendered_diagram_cache.append(full_result) 

82 return full_result 

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._process_source_files(service_node, root_path), 

94 ) 

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

96 

97 def _process_source_files( 

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_src_file 

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

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

107 ) 

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

109 all_rendered_files: typing.Final = executor.map( 

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

111 py_files, 

112 ) 

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

114 

115 def _process_one_file( 

116 self, 

117 service_node: diagram_model.DiagramNode, 

118 one_src_file: pathlib.Path, 

119 /, 

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

121 raw_file_source: typing.Final = one_src_file.read_text() 

122 return tuple( 

123 one_edge 

124 for one_feature_functions in MAPPING_OF_PARSERS_AND_RENDERERS.values() 

125 for one_edge in one_feature_functions.render_diagram( 

126 service_node, 

127 one_feature_functions.parse_source(raw_file_source), 

128 ) 

129 )