Coverage for faststream / _internal / utils / path.py: 83%

1import re (empty)

2from collections.abc import Callable (empty)

3from re import Pattern (empty)

4from typing import Any (empty)

5 

6from faststream.exceptions import SetupError (empty)

7 

8PARAM_REGEX = re.compile(r"{([a-zA-Z0-9_]+)}") (empty)

9 

10 

11def compile_path( (empty)

12 path: str, 

13 replace_symbol: str, 

14 patch_regex: Callable[[str], str] = lambda x: x, 

15) -> tuple[Pattern[str] | None, str]: 

16 path_regex = "^.*?" (empty)

17 original_path = "" (empty)

18 

19 idx = 0 (empty)

20 params = set() (empty)

21 duplicated_params = set() (empty)

22 for match in PARAM_REGEX.finditer(path): (empty)

23 param_name = match.groups("str")[0] (empty)

24 

25 path_regex += re.escape(path[idx : match.start()]) (empty)

26 path_regex += f"(?P<{param_name.replace('+', '')}>[^.]+)" (empty)

27 

28 original_path += path[idx : match.start()] (empty)

29 original_path += replace_symbol (empty)

30 

31 if param_name in params: 31 ↛ 32line 31 didn't jump to line 32 because the condition on line 31 was never true(empty)

32 duplicated_params.add(param_name) 

33 else: 

34 params.add(param_name) (empty)

35 

36 idx = match.end() (empty)

37 

38 if duplicated_params: 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true(empty)

39 names = ", ".join(sorted(duplicated_params)) 

40 ending = "s" if len(duplicated_params) > 1 else "" 

41 msg = f"Duplicated param name{ending} {names} at path {path}" 

42 raise SetupError(msg) 

43 

44 if idx == 0: (empty)

45 regex = None (empty)

46 else: 

47 path_regex += re.escape(path[idx:]) + "$" (empty)

48 regex = re.compile(patch_regex(path_regex)) (empty)

49 

50 original_path += path[idx:] (empty)

51 return regex, original_path (empty)

52 

53 

54def match_path(pattern: Pattern[str] | None, subject: str) -> dict[str, Any]: (empty)

55 """Match subject against pattern and return named groups, or {} if no match.""" 

56 if pattern is not None and (match := pattern.match(subject)): (empty)

57 return match.groupdict() (empty)

58 return {} (empty)