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

1import re as py_re 

2import types 

3import typing 

4 

5from archdocs import prefilter, settings 

6from archdocs.features.messaging_queue import const 

7 

8 

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

10# The `\b` is not decoration: without it every position inside a long word run — a base64 blob 

11# in a string is enough — starts its own `\w+` attempt, and one file costs seconds instead of 

12# microseconds. 

13_PRODUCER_PATTERN: typing.Final = py_re.compile( 

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

15 flags=settings.TYPICAL_RE_FLAGS, 

16) 

17_BROKER_PATTERNS: typing.Final = types.MappingProxyType( 

18 { 

19 one_broker: py_re.compile( 

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

21 flags=settings.TYPICAL_RE_FLAGS, 

22 ) 

23 for one_broker in const.BrokerEnum 

24 }, 

25) 

26# A broker assignment always opens its line, and the anchor is what caps the cost: unanchored, 

27# every position of a long word run starts a `\w+` attempt that backtracks hunting for `=`. 

28_BROKER_VARIABLE_PATTERN: typing.Final = py_re.compile( 

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

30 flags=settings.TYPICAL_RE_FLAGS, 

31) 

32_TOPIC_PATTERNS_OF_DIRECTION: typing.Final = types.MappingProxyType( 

33 { 

34 const.MessageDirection.consumed: ( 

35 py_re.compile( 

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

37 flags=settings.TYPICAL_RE_FLAGS, 

38 ), 

39 ), 

40 const.MessageDirection.produced: ( 

41 py_re.compile( 

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

43 flags=settings.TYPICAL_RE_FLAGS, 

44 ), 

45 py_re.compile( 

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

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

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

49 flags=settings.TYPICAL_RE_FLAGS, 

50 ), 

51 ), 

52 }, 

53) 

54_BROKER_NAME_OF_CLASS: typing.Final = types.MappingProxyType( 

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

56) 

57_FASTSTREAM_LITERALS: typing.Final = ("faststream",) 

58_EMPTY_FEATURES: typing.Final = const.MessagingQueueFeatures() 

59 

60 

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

62 return { 

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

64 for one_match in _BROKER_VARIABLE_PATTERN.finditer(raw_source) 

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

66 } 

67 

68 

69def _collect_topics_of_broker( 

70 raw_source: str, 

71 broker_of_variable: typing.Mapping[str, str], 

72 message_direction: const.MessageDirection, 

73 broker_name: str, 

74 /, 

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

76 return tuple( 

77 dict.fromkeys( 

78 one_match.group("topic") 

79 for one_pattern in _TOPIC_PATTERNS_OF_DIRECTION[message_direction] 

80 for one_match in one_pattern.finditer(raw_source) 

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

82 ), 

83 ) 

84 

85 

86def _build_broker_flow( 

87 raw_source: str, 

88 broker_of_variable: typing.Mapping[str, str], 

89 broker_name: str, 

90 /, 

91) -> const.BrokerFlow: 

92 return const.BrokerFlow( 

93 broker_name=broker_name, 

94 consumes=bool(_SUBSCRIBER_DECORATOR_PATTERN.search(raw_source)), 

95 produces=bool(_PRODUCER_PATTERN.search(raw_source)), 

96 consumed_topics=_collect_topics_of_broker( 

97 raw_source, 

98 broker_of_variable, 

99 const.MessageDirection.consumed, 

100 broker_name, 

101 ), 

102 produced_topics=_collect_topics_of_broker( 

103 raw_source, 

104 broker_of_variable, 

105 const.MessageDirection.produced, 

106 broker_name, 

107 ), 

108 ) 

109 

110 

111def find_faststream_features(raw_source: str) -> const.MessagingQueueFeatures: 

112 if not prefilter.contains_any_literal(raw_source.lower(), _FASTSTREAM_LITERALS): 

113 return _EMPTY_FEATURES 

114 if not _SUBSCRIBER_DECORATOR_PATTERN.search(raw_source) and not _PRODUCER_PATTERN.search(raw_source): 

115 return _EMPTY_FEATURES 

116 broker_of_variable: typing.Final = _collect_broker_of_variable(raw_source) 

117 return const.MessagingQueueFeatures( 

118 broker_flows=tuple( 

119 _build_broker_flow(raw_source, broker_of_variable, one_broker.value) 

120 for one_broker, one_pattern in _BROKER_PATTERNS.items() 

121 if one_pattern.search(raw_source) 

122 ), 

123 )