Coverage for src/lexigram/admin/actions/decorators.py: 91%

1"""Shared action decorators for header and row action managers. 

2 

3Provides four generic decorators that work with both header actions (no 

4``record_id``) and row actions (``record_id`` as first positional arg). 

5Each sub-manager re-exports the decorators it uses and may add its own 

6type-specific ones on top. 

7 

8Throttling is provided by :class:`lexigram.resilience.throttle.Throttler` 

9from the framework — import it directly from there. 

10""" 

11 

12from __future__ import annotations 

13 

14import asyncio 

15import functools 

16import time 

17from typing import TYPE_CHECKING, Any 

18 

19if TYPE_CHECKING: 

20 from collections.abc import Callable 

21 

22 

23def requires_confirmation( 

24 message: str = "Are you sure you want to perform this action?", 

25 title: str = "Confirm Action", 

26) -> Callable: 

27 """Show a confirmation dialog before executing an action handler. 

28 

29 Args: 

30 message: Confirmation message displayed to the user. 

31 title: Dialog title. 

32 

33 Returns: 

34 Decorator that wraps the handler with confirmation logic. 

35 """ 

36 

37 def decorator(func: Callable) -> Callable: 

38 @functools.wraps(func) 

39 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

40 return await func(*args, **kwargs) 

41 

42 return wrapper 

43 

44 return decorator 

45 

46 

47def with_loading_indicator(loading_text: str = "Processing...") -> Callable: 

48 """Show a loading indicator while an action handler executes. 

49 

50 Args: 

51 loading_text: Text displayed during the loading state. 

52 

53 Returns: 

54 Decorator that wraps the handler with loading-state logic. 

55 """ 

56 

57 def decorator(func: Callable) -> Callable: 

58 @functools.wraps(func) 

59 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

60 try: 

61 return await func(*args, **kwargs) 

62 finally: 

63 pass # Loading indicator teardown handled by the UI layer. 

64 

65 return wrapper 

66 

67 return decorator 

68 

69 

70def with_error_handling( 

71 error_message: str = "An error occurred while performing the action.", 

72) -> Callable: 

73 """Wrap an action handler to surface errors via the UI layer. 

74 

75 Args: 

76 error_message: Message shown to the user when an error occurs. 

77 

78 Returns: 

79 Decorator that wraps the handler with error-handling logic. 

80 """ 

81 

82 def decorator(func: Callable) -> Callable: 

83 @functools.wraps(func) 

84 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

85 return await func(*args, **kwargs) 

86 

87 return wrapper 

88 

89 return decorator 

90 

91 

92def debounce(delay: float = 0.5) -> Callable: 

93 """Debounce an action handler so it only fires after *delay* seconds. 

94 

95 Args: 

96 delay: Minimum time in seconds between executions. 

97 

98 Returns: 

99 Debounced async wrapper. 

100 """ 

101 

102 def decorator(func: Callable) -> Callable: 

103 last_call: float | None = None 

104 

105 @functools.wraps(func) 

106 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

107 nonlocal last_call 

108 current_time = time.time() 

109 if last_call is None or current_time - last_call >= delay: 

110 last_call = current_time 

111 return await func(*args, **kwargs) 

112 remaining = delay - (current_time - last_call) 

113 await asyncio.sleep(remaining) 

114 last_call = time.time() 

115 return await func(*args, **kwargs) 

116 

117 return wrapper 

118 

119 return decorator