Coverage for src/lexigram/web/pipes/decorators.py: 42%

1"""Pipes decorators for Lexigram Framework. 

2 

3Provides @use_pipes for applying pipes to controllers and handlers. 

4""" 

5 

6from __future__ import annotations 

7 

8from collections.abc import Callable 

9import inspect 

10from typing import Any, TypeVar, cast 

11 

12from lexigram.web.protocols import PipeProtocol 

13 

14T = TypeVar("T") 

15 

16 

17def use_pipes(*pipe_instances: PipeProtocol) -> Callable[[T], T]: 

18 """Decorator to apply pipes to a class or method. 

19 

20 Stores pipe metadata on the target for runtime resolution by the ParameterBinder. 

21 Pipes applied at the class level are inherited by all handler methods. 

22 

23 Example:: 

24 

25 @use_pipes(ValidationPipe()) 

26 class UserController(Controller): 

27 @post("/users") 

28 async def create(self, user: User): 

29 ... 

30 

31 Args: 

32 *pipe_instances: PipeProtocol instances to apply. 

33 

34 Returns: 

35 The decorated class or function. 

36 """ 

37 

38 def decorator(target: T) -> T: 

39 if inspect.isclass(target): 

40 # Apply to all methods of the class 

41 for name, method in inspect.getmembers( 

42 target, 

43 predicate=inspect.isfunction, 

44 ): 

45 if not name.startswith("_"): # Skip private methods 

46 existing: list[PipeProtocol] = ( 

47 getattr(method, "_lexigram_pipes", None) or [] 

48 ) 

49 # Prepend class-level pipes so they run before method-level ones 

50 cast("Any", method)._lexigram_pipes = ( 

51 list(pipe_instances) + existing 

52 ) 

53 return cast("T", target) 

54 

55 # Apply to a single function/method 

56 existing_pipes: list[PipeProtocol] = ( 

57 getattr(target, "_lexigram_pipes", None) or [] 

58 ) 

59 cast("Any", target)._lexigram_pipes = existing_pipes + list(pipe_instances) 

60 return target 

61 

62 return decorator 

63 

64 

65__all__ = ["use_pipes"]