Coverage for archdocs/features/messaging_queue/parser.py: 100%

1import re as py_re 

2import types 

3import typing 

4 

5from archdocs import settings 

6from archdocs.features.messaging_queue import const 

7 

8 

9_SUBSCRIBER_DECORATOR_RE: typing.Final = py_re.compile(r"@\w+\.subscriber\(", flags=settings.TYPICAL_RE_FLAGS) 

10_PRODUCER_RE: typing.Final = py_re.compile( 

11 r"(?:@\w+\.(?:publisher|producer)|\w+\.publish)\(", 

12 flags=settings.TYPICAL_RE_FLAGS, 

13) 

14_BROKER_PATTERNS: typing.Final = types.MappingProxyType( 

15 { 

16 one_broker: py_re.compile( 

17 rf"\bfaststream\.{one_broker.value}\b", 

18 flags=settings.TYPICAL_RE_FLAGS, 

19 ) 

20 for one_broker in const.BrokersEnum 

21 }, 

22) 

23_BROKER_VARIABLE_PATTERN: typing.Final = py_re.compile( 

24 r"(?P<variable>\w+)\s*(?::[^=\n]+)?=\s*(?P<broker_class>\w+)\s*\(", 

25 flags=settings.TYPICAL_RE_FLAGS, 

26) 

27_TOPIC_PATTERNS_OF_DIRECTION: typing.Final = types.MappingProxyType( 

28 { 

29 const.MessageDirection.consumed: ( 

30 py_re.compile( 

31 r"@(?P<variable>\w+)\.subscriber\(\s*[\"'](?P<topic>[^\"']+)[\"']", 

32 flags=settings.TYPICAL_RE_FLAGS, 

33 ), 

34 ), 

35 const.MessageDirection.produced: ( 

36 py_re.compile( 

37 r"@(?P<variable>\w+)\.publisher\(\s*[\"'](?P<topic>[^\"']+)[\"']", 

38 flags=settings.TYPICAL_RE_FLAGS, 

39 ), 

40 py_re.compile( 

41 r"(?P<variable>\w+)\.publish\([^()]*?\b(?:" 

42 + "|".join(const.DESTINATION_KEYWORDS) 

43 + r")\s*=\s*[\"'](?P<topic>[^\"']+)[\"']", 

44 flags=settings.TYPICAL_RE_FLAGS, 

45 ), 

46 ), 

47 }, 

48) 

49_BROKER_NAME_OF_CLASS: typing.Final = types.MappingProxyType( 

50 {one_broker_class: one_broker.value for one_broker, one_broker_class in const.BROKER_CLASS_OF_NAME.items()}, 

51) 

52_EMPTY_FEATURES: typing.Final = const.MQFeatures() 

53 

54 

55def _collect_broker_of_variable(raw_source: str, /) -> dict[str, str]: 

56 return { 

57 one_match.group("variable"): _BROKER_NAME_OF_CLASS[one_match.group("broker_class")] 

58 for one_match in _BROKER_VARIABLE_PATTERN.finditer(raw_source) 

59 if one_match.group("broker_class") in _BROKER_NAME_OF_CLASS 

60 } 

61 

62 

63def _collect_topics_of_broker( 

64 raw_source: str, 

65 broker_of_variable: dict[str, str], 

66 message_direction: const.MessageDirection, 

67 broker_name: str, 

68 /, 

69) -> tuple[str, ...]: 

70 return tuple( 

71 dict.fromkeys( 

72 one_match.group("topic") 

73 for one_pattern in _TOPIC_PATTERNS_OF_DIRECTION[message_direction] 

74 for one_match in one_pattern.finditer(raw_source) 

75 if broker_of_variable.get(one_match.group("variable")) == broker_name 

76 ), 

77 ) 

78 

79 

80def _build_broker_flow(raw_source: str, broker_of_variable: dict[str, str], broker_name: str, /) -> const.BrokerFlow: 

81 return const.BrokerFlow( 

82 broker_name=broker_name, 

83 consumes=bool(_SUBSCRIBER_DECORATOR_RE.search(raw_source)), 

84 produces=bool(_PRODUCER_RE.search(raw_source)), 

85 consumed_topics=_collect_topics_of_broker( 

86 raw_source, 

87 broker_of_variable, 

88 const.MessageDirection.consumed, 

89 broker_name, 

90 ), 

91 produced_topics=_collect_topics_of_broker( 

92 raw_source, 

93 broker_of_variable, 

94 const.MessageDirection.produced, 

95 broker_name, 

96 ), 

97 ) 

98 

99 

100def find_faststream_features(raw_source: str) -> const.MQFeatures: 

101 if "faststream" not in raw_source: 

102 return _EMPTY_FEATURES 

103 if not _SUBSCRIBER_DECORATOR_RE.search(raw_source) and not _PRODUCER_RE.search(raw_source): 

104 return _EMPTY_FEATURES 

105 broker_of_variable: typing.Final = _collect_broker_of_variable(raw_source) 

106 return const.MQFeatures( 

107 broker_flows=tuple( 

108 _build_broker_flow(raw_source, broker_of_variable, one_broker.value) 

109 for one_broker, one_pattern in _BROKER_PATTERNS.items() 

110 if one_pattern.search(raw_source) 

111 ), 

112 )