Coverage for nexios\decorators.py: 56%

1from typing import List, Dict, Any, TypeVar, Union, Callable, Type, Tuple 

2from .http.request import Request 

3from .http.response import NexiosResponse 

4from .http.request import Request 

5import typing 

6from functools import wraps 

7from .types import HandlerType 

8 

9F = TypeVar("F", bound=HandlerType) 

10 

11 

12class RouteDecorator: 

13 """Base class for all route decorators""" 

14 

15 def __init__(self, **kwargs: Dict[str, Any]): 

16 pass 

17 

18 def __call__(self, handler: HandlerType) -> Any: 

19 raise NotImplementedError("Handler not set") 

20 

21 def __get__(self, obj: typing.Any, objtype: typing.Any = None): 

22 if obj is None: 

23 return self 

24 return self.__class__(obj) # type:ignore 

25 

26 

27class allowed_methods(RouteDecorator): 

28 def __init__(self, methods: List[str]) -> None: 

29 super().__init__() 

30 self.allowed_methods: List[str] = [method.upper() for method in methods] 

31 

32 def __call__(self, handler: F) -> F: 

33 if getattr(handler, "_is_wrapped", False): 

34 return handler 

35 

36 @wraps(handler) 

37 async def wrapper(*args: List[Any], **kwargs: Dict[str, Any]) -> Any: 

38 *_, request, response = args # Ensure request and response are last 

39 

40 if not isinstance(request, Request) or not isinstance( 

41 response, NexiosResponse 

42 ): 

43 raise TypeError("Expected request and response as the first arguments") 

44 

45 if request.method.upper() not in self.allowed_methods: 

46 return response.json( 

47 { 

48 "error": f"Method {request.method} not allowed", 

49 "allowed_methods": self.allowed_methods, 

50 }, 

51 status_code=405, 

52 headers={"Allow": ", ".join(self.allowed_methods)}, 

53 ) 

54 

55 return await handler(*args, **kwargs) # type:ignore 

56 

57 wrapper._is_wrapped = True # type: ignore 

58 return wrapper # type: ignore 

59 

60 

61class catch_exception(RouteDecorator): 

62 """Decorator to catch specific exceptions and handle them with a custom handler""" 

63 

64 def __init__( 

65 self, 

66 exceptions: Union[Type[Exception], Tuple[Type[Exception], ...]], 

67 handler: Callable[[Exception, Request, NexiosResponse], Any], 

68 ) -> None: 

69 super().__init__() 

70 if not isinstance(exceptions, tuple): 

71 exceptions = (exceptions,) 

72 self.exceptions = exceptions 

73 self.exception_handler = handler 

74 

75 def __call__(self, handler: F) -> F: 

76 if getattr(handler, "_is_wrapped", False): 

77 return handler 

78 

79 @wraps(handler) 

80 async def wrapper(*args: List[Any], **kwargs: Dict[str, Any]) -> Any: 

81 try: 

82 return await handler(*args, **kwargs) # type: ignore 

83 except self.exceptions as e: 

84 *_, request, response = args 

85 

86 if not isinstance(request, Request) or not isinstance( 

87 response, NexiosResponse 

88 ): 

89 raise TypeError( 

90 "Expected request and response as the last arguments" 

91 ) 

92 

93 return self.exception_handler(request, response, e) # type:ignore 

94 

95 wrapper._is_wrapped = True # type: ignore 

96 return wrapper # type: ignore