Coverage for lintro / utils / output / parser_registry.py: 100%

1"""Registry for tool output parsers and fixability predicates. 

2 

3This module provides a registry pattern for O(1) lookup of tool parsers 

4and fixability predicates, replacing the O(n) if/elif chains. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass 

10from typing import TYPE_CHECKING, Any 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Callable, Sequence 

14 

15 ParserFunc = Callable[[str], Sequence[Any]] 

16 FixabilityPredicate = Callable[[object], bool] 

17 

18 

19@dataclass(frozen=True) 

20class ParserEntry: 

21 """Registry entry for a tool parser. 

22 

23 Attributes: 

24 parse_func: Function to parse tool output into issues. 

25 is_fixable: Optional predicate to determine if an issue is fixable. 

26 """ 

27 

28 parse_func: ParserFunc 

29 is_fixable: FixabilityPredicate | None = None 

30 

31 

32class ParserRegistry: 

33 """O(1) lookup registry for tool output parsers. 

34 

35 This registry stores parser functions and fixability predicates for each 

36 tool, enabling efficient dispatch without long if/elif chains. 

37 

38 Example: 

39 >>> ParserRegistry.register("ruff", parse_ruff_output, is_fixable=ruff_fixable) 

40 >>> issues = ParserRegistry.parse("ruff", output) 

41 """ 

42 

43 _parsers: dict[str, ParserEntry] = {} 

44 

45 @classmethod 

46 def register( 

47 cls, 

48 tool_name: str, 

49 parse_func: ParserFunc, 

50 is_fixable: FixabilityPredicate | None = None, 

51 ) -> None: 

52 """Register a parser for a tool. 

53 

54 Args: 

55 tool_name: Name of the tool (case-insensitive). 

56 parse_func: Function that parses tool output into issues. 

57 is_fixable: Optional predicate to check if an issue is fixable. 

58 """ 

59 cls._parsers[tool_name.lower()] = ParserEntry( 

60 parse_func=parse_func, 

61 is_fixable=is_fixable, 

62 ) 

63 

64 @classmethod 

65 def get(cls, tool_name: str) -> ParserEntry | None: 

66 """Get parser entry for a tool. 

67 

68 Args: 

69 tool_name: Name of the tool (case-insensitive). 

70 

71 Returns: 

72 ParserEntry if registered, None otherwise. 

73 """ 

74 return cls._parsers.get(tool_name.lower()) 

75 

76 @classmethod 

77 def parse(cls, tool_name: str, output: str) -> list[Any]: 

78 """Parse output using registered parser. 

79 

80 Args: 

81 tool_name: Name of the tool (case-insensitive). 

82 output: Raw output string from the tool. 

83 

84 Returns: 

85 List of parsed issues, or empty list if no parser registered. 

86 """ 

87 entry = cls.get(tool_name) 

88 if entry is None: 

89 return [] 

90 return list(entry.parse_func(output)) 

91 

92 @classmethod 

93 def get_fixability_predicate( 

94 cls, 

95 tool_name: str, 

96 ) -> FixabilityPredicate | None: 

97 """Get fixability predicate for a tool. 

98 

99 Args: 

100 tool_name: Name of the tool (case-insensitive). 

101 

102 Returns: 

103 Fixability predicate function, or None if not registered. 

104 """ 

105 entry = cls.get(tool_name) 

106 return entry.is_fixable if entry else None 

107 

108 @classmethod 

109 def clear(cls) -> None: 

110 """Clear all registered parsers. 

111 

112 Primarily useful for testing to reset registry state. 

113 """ 

114 cls._parsers = {} 

115 

116 @classmethod 

117 def is_registered(cls, tool_name: str) -> bool: 

118 """Check if a tool has a registered parser. 

119 

120 Args: 

121 tool_name: Name of the tool (case-insensitive). 

122 

123 Returns: 

124 True if the tool has a registered parser. 

125 """ 

126 return tool_name.lower() in cls._parsers