Coverage for src/lexigram/admin/actions/polymorphic.py: 0%

1"""PolymorphicBulkAction — dispatches bulk records to per-type action handlers. 

2 

3.. stability:: stable 

4 

5Usage:: 

6 

7 class MarkReviewedBulkAction(PolymorphicBulkAction): 

8 name = "mark_reviewed" 

9 handlers: ClassVar[dict[type, Action[Any, Any]]] = { 

10 AIAnalysis: MarkAIAnalysisReviewedAction(name="mark_reviewed"), 

11 VetSubmission: MarkVetSubmissionReviewedAction(name="mark_reviewed"), 

12 } 

13 

14The ``handlers`` map is a ``ClassVar`` (not a dataclass field) so that 

15``dict`` values are not subject to frozen-dataclass hash requirements. 

16First ``isinstance`` match wins for each record. 

17""" 

18 

19from __future__ import annotations 

20 

21from typing import Any, ClassVar 

22 

23from lexigram.admin.actions.base import Action, BulkAction 

24from lexigram.admin.actions.exceptions import ActionError 

25from lexigram.admin.actions.types import ActionContext 

26from lexigram.result import Err, Ok, Result 

27 

28 

29class PolymorphicBulkAction(BulkAction): 

30 """Bulk action that dispatches records to per-type handler instances. 

31 

32 Subclasses declare a ``handlers`` class variable mapping record types 

33 to ``Action`` instances. On ``execute()``, each record in the input 

34 list is matched against the handlers map via ``isinstance`` (first 

35 match wins). Records that match no handler cause the action to return 

36 ``Err``. If any handler returns ``Err``, the overall result is ``Err``. 

37 

38 Attributes: 

39 handlers: Class-level mapping of record type → Action handler. 

40 """ 

41 

42 handlers: ClassVar[dict[type, Action[Any, Any]]] = {} 

43 

44 async def execute( 

45 self, 

46 record_or_records: list[Any], 

47 ctx: ActionContext, 

48 ) -> Result[Any, ActionError]: 

49 outcomes: list[Any] = [] 

50 

51 for record in record_or_records: 

52 handler: Action[Any, Any] | None = None 

53 for record_type, candidate in self.handlers.items(): 

54 if isinstance(record, record_type): 

55 handler = candidate 

56 break 

57 

58 if handler is None: 

59 return Err( 

60 ActionError( 

61 f"No handler registered for record type " 

62 f"'{type(record).__name__}' in {type(self).__name__}" 

63 ) 

64 ) 

65 

66 result = await handler.execute(record, ctx) 

67 if result.is_err(): 

68 return result 

69 outcomes.append(result.unwrap()) 

70 

71 return Ok(outcomes) 

72 

73 

74__all__ = ["PolymorphicBulkAction"]