Coverage for src/lexigram/admin/integrations/tasks.py: 80%

1"""Tasks integration — dispatches bulk actions through a task queue.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.core.di import ( 

9 ContainerRegistrarProtocol, 

10 ContainerResolverProtocol, 

11 ) 

12 

13 

14class _NoOpTasks: 

15 async def dispatch( 

16 self, runner: str, action_name: str, record_ids: list[str], ctx_summary: str 

17 ) -> dict[str, Any]: 

18 return {"status": "noop"} 

19 

20 

21class TasksIntegration: 

22 """Adapter that dispatches bulk actions via lexigram-tasks. 

23 

24 Gracefully no-ops when ``lexigram-tasks`` is not installed or the 

25 integration is disabled. 

26 """ 

27 

28 def __init__(self, config: Any) -> None: 

29 self._config = config 

30 self._tasks: Any = None 

31 self._enabled = False 

32 

33 def register(self, container: ContainerRegistrarProtocol) -> None: 

34 from lexigram.admin.config import TasksIntegrationConfig 

35 from lexigram.admin.integrations._optional import is_installed 

36 

37 cfg = self._config 

38 if not isinstance(cfg, TasksIntegrationConfig): 

39 cfg = TasksIntegrationConfig() 

40 if not cfg.enabled: 

41 self._tasks = _NoOpTasks() 

42 return 

43 if not is_installed("lexigram.tasks"): 

44 self._tasks = _NoOpTasks() 

45 return 

46 self._enabled = True 

47 

48 async def boot(self, container: ContainerResolverProtocol) -> None: 

49 if not self._enabled: 

50 return 

51 try: 

52 from lexigram.contracts.infra.tasks import TaskQueueProtocol 

53 

54 self._tasks = await container.resolve(TaskQueueProtocol) 

55 except Exception: # noqa: BLE001 

56 self._tasks = _NoOpTasks() 

57 

58 async def shutdown(self) -> None: 

59 pass 

60 

61 async def health_check(self) -> dict[str, Any]: 

62 return { 

63 "status": "healthy" if not isinstance(self._tasks, _NoOpTasks) else "noop" 

64 } 

65 

66 async def dispatch( 

67 self, 

68 runner: str, 

69 action_name: str, 

70 record_ids: list[str], 

71 ctx_summary: str, 

72 ) -> dict[str, Any]: 

73 return await self._tasks.dispatch(runner, action_name, record_ids, ctx_summary) 

74 

75 @property 

76 def threshold(self) -> int: 

77 return getattr(self._config, "bulk_threshold", 25) 

78 

79 

80__all__ = ["TasksIntegration"]