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 httpapi_parser 

11from archdocs.features.http_api import renderer as httpapi_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 mq_parser 

18from archdocs.features.messaging_queue import renderer as mq_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@typing.final 

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

29class _FeatureFunctions: 

30 parse_source: typing.Callable[[str], typing.Any] 

31 render_diagram: typing.Callable[[diagram_model.DiagramNode, typing.Any], tuple[diagram_model.DiagramEdge, ...]] 

32 

33 

34@typing.final 

35class AllCurrentFeatures(enum.Enum): 

36 fastapi_litestar = 1 

37 faststream = 2 

38 sqlalchemy = 3 

39 redis_cache = 4 

40 http_clients = 5 

41 task_queues = 6 

42 app_servers = 7 

43 

44 

45MAPPING_OF_PARSERS_AND_RENDERERS: typing.Final = types.MappingProxyType( 

46 { 

47 AllCurrentFeatures.fastapi_litestar: _FeatureFunctions( 

48 parse_source=httpapi_parser.find_fastapi_and_litestar_features, 

49 render_diagram=httpapi_renderer.render_http_api_features, 

50 ), 

51 AllCurrentFeatures.faststream: _FeatureFunctions( 

52 parse_source=mq_parser.find_faststream_features, 

53 render_diagram=mq_renderer.render_mq_features, 

54 ), 

55 AllCurrentFeatures.sqlalchemy: _FeatureFunctions( 

56 parse_source=sqlalchemy_parser.find_sqlalchemy_features, 

57 render_diagram=sqlalchemy_renderer.render_sqlalchemy_features, 

58 ), 

59 AllCurrentFeatures.redis_cache: _FeatureFunctions( 

60 parse_source=redis_parser.find_redis_features, 

61 render_diagram=redis_renderer.render_redis_features, 

62 ), 

63 AllCurrentFeatures.http_clients: _FeatureFunctions( 

64 parse_source=http_clients_parser.find_http_client_features, 

65 render_diagram=http_clients_renderer.render_http_client_features, 

66 ), 

67 AllCurrentFeatures.task_queues: _FeatureFunctions( 

68 parse_source=task_queues_parser.find_task_queue_features, 

69 render_diagram=task_queues_renderer.render_task_queue_features, 

70 ), 

71 AllCurrentFeatures.app_servers: _FeatureFunctions( 

72 parse_source=app_servers_parser.find_application_server_features, 

73 render_diagram=app_servers_renderer.render_application_server_features, 

74 ), 

75 }, 

76) 

77 

78 

79@typing.final 

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

81class ManifestFeatureFunctions: 

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

83 parse_manifests: typing.Callable[[str], typing.Any] 

84 render_diagram: typing.Callable[[diagram_model.DiagramNode, typing.Any], tuple[diagram_model.DiagramEdge, ...]] 

85 render_node_annotations: typing.Callable[[typing.Any], tuple[str, ...]] 

86 

87 

88@typing.final 

89class AllCurrentManifestFeatures(enum.Enum): 

90 kubernetes = 1 

91 

92 

93MAPPING_OF_MANIFEST_PARSERS_AND_RENDERERS: typing.Final = types.MappingProxyType( 

94 { 

95 AllCurrentManifestFeatures.kubernetes: ManifestFeatureFunctions( 

96 read_source=kubernetes_manifest_files.read_kubernetes_manifests, 

97 parse_manifests=kubernetes_parser.find_kubernetes_features, 

98 render_diagram=kubernetes_renderer.render_kubernetes_features, 

99 render_node_annotations=kubernetes_renderer.render_kubernetes_node_annotations, 

100 ), 

101 }, 

102)