Coverage for src/lexigram/web/pipes/pipeline.py: 34%

1"""PipeProtocol pipeline for running pipes on parameters. 

2 

3The pipeline runs pipes in order, transforming each parameter value. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING, Any 

9 

10if TYPE_CHECKING: 

11 from lexigram.web.protocols import ParamMetadata, PipeProtocol 

12 

13 

14class PipePipeline: 

15 """Pipeline that runs pipes on handler parameters. 

16 

17 Processes parameters through registered pipes in order, transforming 

18 and validating values before they reach the handler. 

19 """ 

20 

21 def __init__(self, pipes: dict[str, list[PipeProtocol]] | None = None): 

22 """Initialize the pipe pipeline. 

23 

24 Args: 

25 pipes: Optional dict mapping parameter names to lists of pipes. 

26 """ 

27 self._pipes: dict[str, list[PipeProtocol]] = pipes or {} 

28 

29 def add_pipe(self, param_name: str, pipe: PipeProtocol) -> None: 

30 """Add a pipe for a parameter. 

31 

32 Args: 

33 param_name: The name of the parameter. 

34 pipe: The pipe to add. 

35 """ 

36 if param_name not in self._pipes: 

37 self._pipes[param_name] = [] 

38 if pipe not in self._pipes[param_name]: 

39 self._pipes[param_name].append(pipe) 

40 

41 def remove_pipe(self, param_name: str, pipe: PipeProtocol) -> None: 

42 """Remove a pipe for a parameter. 

43 

44 Args: 

45 param_name: The name of the parameter. 

46 pipe: The pipe to remove. 

47 """ 

48 if param_name in self._pipes: 

49 if pipe in self._pipes[param_name]: 

50 self._pipes[param_name].remove(pipe) 

51 if not self._pipes[param_name]: 

52 del self._pipes[param_name] 

53 

54 def get_pipes(self, param_name: str) -> list[PipeProtocol]: 

55 """Get pipes for a parameter. 

56 

57 Args: 

58 param_name: The name of the parameter. 

59 

60 Returns: 

61 List of pipes for the parameter. 

62 """ 

63 return list(self._pipes.get(param_name, [])) 

64 

65 async def transform( 

66 self, 

67 param_name: str, 

68 value: Any, 

69 metadata: ParamMetadata, 

70 ) -> Any: 

71 """Transform a parameter value through its pipes. 

72 

73 Args: 

74 param_name: The name of the parameter. 

75 value: The value to transform. 

76 metadata: Metadata about the parameter. 

77 

78 Returns: 

79 Transformed value. 

80 """ 

81 pipes = self.get_pipes(param_name) 

82 

83 if not pipes: 

84 return value 

85 

86 result = value 

87 for pipe in pipes: 

88 result = await pipe.transform(result, metadata) 

89 

90 return result 

91 

92 def __len__(self) -> int: 

93 """Return total number of pipes.""" 

94 return sum(len(p) for p in self._pipes.values()) 

95 

96 

97__all__ = ["PipePipeline"]