Coverage for archdocs/mapping.py: 100%

1import dataclasses 

2import enum 

3import pathlib 

4import types 

5import typing 

6 

7from archdocs import diagram_model 

8from archdocs.features.app_servers import parser as app_servers_parser 

9from archdocs.features.app_servers import renderer as app_servers_renderer 

10from archdocs.features.http_api import parser as http_api_parser 

11from archdocs.features.http_api import renderer as http_api_renderer 

12from archdocs.features.http_clients import parser as http_clients_parser 

13from archdocs.features.http_clients import renderer as http_clients_renderer 

14from archdocs.features.kubernetes import manifest_files as kubernetes_manifest_files 

15from archdocs.features.kubernetes import parser as kubernetes_parser 

16from archdocs.features.kubernetes import renderer as kubernetes_renderer 

17from archdocs.features.messaging_queue import parser as messaging_queue_parser 

18from archdocs.features.messaging_queue import renderer as messaging_queue_renderer 

19from archdocs.features.redis import parser as redis_parser 

20from archdocs.features.redis import renderer as redis_renderer 

21from archdocs.features.sqlalchemy import parser as sqlalchemy_parser 

22from archdocs.features.sqlalchemy import renderer as sqlalchemy_renderer 

23from archdocs.features.task_queues import parser as task_queues_parser 

24from archdocs.features.task_queues import renderer as task_queues_renderer 

25 

26 

27# The type variable is the registry's safety net: an entry pairing one feature's parser with 

28# another feature's renderer — the realistic slip when a new feature is copied from a neighbour — 

29# fails right here under mypy instead of failing at runtime on somebody's diagram. 

30@typing.final 

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

32class _FeatureFunctions[ParsedFeaturesT]: 

33 parse_source: typing.Callable[[str], ParsedFeaturesT] 

34 render_edges: typing.Callable[ 

35 [diagram_model.DiagramNode, ParsedFeaturesT], 

36 tuple[diagram_model.DiagramEdge, ...], 

37 ] 

38 

39 

40@typing.final 

41class AllCurrentFeatures(enum.Enum): 

42 fastapi_litestar = 1 

43 faststream = 2 

44 sqlalchemy = 3 

45 redis_cache = 4 

46 http_clients = 5 

47 task_queues = 6 

48 app_servers = 7 

49 

50 

51type _FeatureRegistry = types.MappingProxyType[AllCurrentFeatures, _FeatureFunctions[typing.Any]] 

52 

53MAPPING_OF_PARSERS_AND_RENDERERS: typing.Final[_FeatureRegistry] = types.MappingProxyType( 

54 { 

55 AllCurrentFeatures.fastapi_litestar: _FeatureFunctions( 

56 parse_source=http_api_parser.find_fastapi_and_litestar_features, 

57 render_edges=http_api_renderer.render_http_api_features, 

58 ), 

59 AllCurrentFeatures.faststream: _FeatureFunctions( 

60 parse_source=messaging_queue_parser.find_faststream_features, 

61 render_edges=messaging_queue_renderer.render_messaging_queue_features, 

62 ), 

63 AllCurrentFeatures.sqlalchemy: _FeatureFunctions( 

64 parse_source=sqlalchemy_parser.find_sqlalchemy_features, 

65 render_edges=sqlalchemy_renderer.render_sqlalchemy_features, 

66 ), 

67 AllCurrentFeatures.redis_cache: _FeatureFunctions( 

68 parse_source=redis_parser.find_redis_features, 

69 render_edges=redis_renderer.render_redis_features, 

70 ), 

71 AllCurrentFeatures.http_clients: _FeatureFunctions( 

72 parse_source=http_clients_parser.find_http_client_features, 

73 render_edges=http_clients_renderer.render_http_client_features, 

74 ), 

75 AllCurrentFeatures.task_queues: _FeatureFunctions( 

76 parse_source=task_queues_parser.find_task_queue_features, 

77 render_edges=task_queues_renderer.render_task_queue_features, 

78 ), 

79 AllCurrentFeatures.app_servers: _FeatureFunctions( 

80 parse_source=app_servers_parser.find_application_server_features, 

81 render_edges=app_servers_renderer.render_application_server_features, 

82 ), 

83 }, 

84) 

85 

86 

87@typing.final 

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

89class ManifestFeatureFunctions[ParsedManifestsT]: 

90 read_source: typing.Callable[[pathlib.Path, str | pathlib.Path | None], str] 

91 parse_manifests: typing.Callable[[str], ParsedManifestsT] 

92 render_edges: typing.Callable[ 

93 [diagram_model.DiagramNode, ParsedManifestsT], 

94 tuple[diagram_model.DiagramEdge, ...], 

95 ] 

96 render_node_annotations: typing.Callable[[ParsedManifestsT], tuple[str, ...]] 

97 

98 

99@typing.final 

100class AllCurrentManifestFeatures(enum.Enum): 

101 kubernetes = 1 

102 

103 

104type _ManifestRegistry = types.MappingProxyType[AllCurrentManifestFeatures, ManifestFeatureFunctions[typing.Any]] 

105 

106MAPPING_OF_MANIFEST_PARSERS_AND_RENDERERS: typing.Final[_ManifestRegistry] = types.MappingProxyType( 

107 { 

108 AllCurrentManifestFeatures.kubernetes: ManifestFeatureFunctions( 

109 read_source=kubernetes_manifest_files.read_kubernetes_manifests, 

110 parse_manifests=kubernetes_parser.find_kubernetes_features, 

111 render_edges=kubernetes_renderer.render_kubernetes_features, 

112 render_node_annotations=kubernetes_renderer.render_kubernetes_node_annotations, 

113 ), 

114 }, 

115)