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

1import re as py_re 

2import types 

3import typing 

4 

5from archdocs import prefilter, settings 

6from archdocs.features.redis.const import RedisConnectionKind, RedisFeatures 

7 

8 

9_REDIS_IMPORT_PATTERN: typing.Final = py_re.compile( 

10 r"\b(?:from\s+redis\b|import\s+redis\b)", 

11 flags=settings.TYPICAL_RE_FLAGS, 

12) 

13_ASYNC_REDIS_PATTERN: typing.Final = py_re.compile( 

14 r"\b(?:from\s+redis\.asyncio\b|import\s+redis\.asyncio\b)", 

15 flags=settings.TYPICAL_RE_FLAGS, 

16) 

17type _ConnectionPatterns = types.MappingProxyType[RedisConnectionKind, py_re.Pattern[str]] 

18 

19# Declaration order is probing order, so the specific kinds go before plain: a file importing 

20# both a Sentinel and a plain Redis client is drawn by the topology, not by the fallback. 

21_REDIS_CONNECTION_PATTERNS: typing.Final[_ConnectionPatterns] = types.MappingProxyType( 

22 { 

23 "sentinel": py_re.compile( 

24 r"\b(?:redis\.(?:asyncio\.)?sentinel\.|from\s+redis(?:\.asyncio)?(?:\.sentinel)?\s+import\s+)" 

25 r".*\bSentinel\b", 

26 flags=settings.TYPICAL_RE_FLAGS, 

27 ), 

28 "cluster": py_re.compile( 

29 r"\b(?:redis\.(?:asyncio\.)?cluster\.|from\s+redis(?:\.asyncio)?(?:\.cluster)?\s+import\s+)" 

30 r".*\bRedisCluster\b", 

31 flags=settings.TYPICAL_RE_FLAGS, 

32 ), 

33 "plain": py_re.compile(r"\b(?:redis\.|from\s+redis\s+import\s+).*\bRedis\b", flags=settings.TYPICAL_RE_FLAGS), 

34 }, 

35) 

36_REDIS_RETRY_PATTERN: typing.Final = py_re.compile( 

37 r"\bredis\.Retry\s*\(", 

38 flags=settings.TYPICAL_RE_FLAGS, 

39) 

40_EMPTY_FEATURES: typing.Final = RedisFeatures( 

41 connection_type=None, 

42 async_used=False, 

43 retry_used=False, 

44) 

45 

46 

47def find_redis_features(raw_source: str) -> RedisFeatures: 

48 if not prefilter.contains_any_literal(raw_source.lower(), ("redis",)): 

49 return _EMPTY_FEATURES 

50 if not _REDIS_IMPORT_PATTERN.search(raw_source): 

51 return _EMPTY_FEATURES 

52 connection_type: typing.Final = next( 

53 ( 

54 one_type_name 

55 for one_type_name, one_pattern in _REDIS_CONNECTION_PATTERNS.items() 

56 if one_pattern.search(raw_source) 

57 ), 

58 None, 

59 ) 

60 return RedisFeatures( 

61 connection_type=connection_type, 

62 async_used=bool(_ASYNC_REDIS_PATTERN.search(raw_source)), 

63 retry_used=bool(_REDIS_RETRY_PATTERN.search(raw_source)), 

64 )