Coverage for src/lexigram/auth/authz/_parsers.py: 77%

1"""Value parser classes for authorization service.""" 

2 

3from __future__ import annotations 

4 

5import threading 

6from typing import Any, Protocol, cast 

7 

8from lexigram import serialization as json 

9 

10 

11class ValueParser(Protocol): 

12 """Protocol for value parsers.""" 

13 

14 def can_parse(self, val: Any) -> bool: 

15 """Check if this parser can handle the given value.""" 

16 ... 

17 

18 def parse(self, val: Any) -> list[str]: 

19 """Parse the value into a list of strings.""" 

20 ... 

21 

22 

23class StringValueParser: 

24 """Parser for string values, including JSON-encoded lists.""" 

25 

26 def can_parse(self, val: Any) -> bool: 

27 return isinstance(val, str) 

28 

29 def parse(self, val: Any) -> list[str]: 

30 try: 

31 parsed = json.loads(val) 

32 return parsed if isinstance(parsed, list) else [str(parsed)] 

33 except (ValueError, TypeError, json.JSONDecodeError): 

34 return [val] 

35 

36 

37class ListValueParser: 

38 """Parser for list values.""" 

39 

40 def can_parse(self, val: Any) -> bool: 

41 return isinstance(val, list) 

42 

43 def parse(self, val: Any) -> list[str]: 

44 return cast("list[str]", val) 

45 

46 

47class NoneValueParser: 

48 """Parser for None values.""" 

49 

50 def can_parse(self, val: Any) -> bool: 

51 return val is None 

52 

53 def parse(self, val: Any) -> list[str]: 

54 return [] 

55 

56 

57class ValueParserRegistry: 

58 """Registry for value parsers. 

59 

60 Provides extensible parsing of various value types into string lists. 

61 Thread-safe registry for value parsers. 

62 """ 

63 

64 def __init__(self) -> None: 

65 self._lock = threading.Lock() 

66 self._parsers: list[ValueParser] = [] 

67 self._register_default_parsers() 

68 

69 def _register_default_parsers(self) -> None: 

70 """Register the default value parsers.""" 

71 self._parsers = [ 

72 StringValueParser(), 

73 ListValueParser(), 

74 NoneValueParser(), 

75 ] 

76 

77 def register_parser(self, parser: ValueParser) -> None: 

78 """Register a custom value parser.""" 

79 with self._lock: 

80 self._parsers.insert(0, parser) 

81 

82 def parse(self, val: Any) -> list[str]: 

83 """Parse the value using registered parsers.""" 

84 with self._lock: 

85 for parser in self._parsers: 

86 if parser.can_parse(val): 

87 return parser.parse(val) 

88 return [str(val)] 

89 

90 

91# Global registry instance 

92_value_parser_registry = ValueParserRegistry() 

93 

94__all__ = [ 

95 "ListValueParser", 

96 "NoneValueParser", 

97 "StringValueParser", 

98 "ValueParser", 

99 "ValueParserRegistry", 

100]