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

1"""AuditedAction — framework base class that wraps execute() with audit logging. 

2 

3Subclasses extend both AuditedAction and a concrete Action subclass 

4(RowAction, BulkAction, HeaderAction). The abstract method is 

5execute_audited(); the inherited execute() is sealed here. 

6 

7Writes are made through ctx.audit_writer. If audit_writer is None, 

8execution proceeds and a warning is logged (best-effort at framework level). 

9""" 

10 

11from __future__ import annotations 

12 

13from abc import abstractmethod 

14from typing import Any, Generic, TypeVar 

15 

16from lexigram.admin.actions.base import Action 

17from lexigram.admin.actions.exceptions import ActionError 

18from lexigram.admin.actions.types import ActionContext 

19from lexigram.contracts.admin.audit_entry import AuditEntry, AuditOutcome 

20from lexigram.logging import get_logger 

21from lexigram.result import Result 

22 

23logger = get_logger(__name__) 

24 

25_RT = TypeVar("_RT") 

26_OT = TypeVar("_OT") 

27 

28 

29class AuditedAction(Action[_RT, _OT], Generic[_RT, _OT]): 

30 """Action base that wraps execute() with before/after audit capture. 

31 

32 Subclass together with a concrete Action specialisation: 

33 

34 class DeleteUser(AuditedAction[User, None], RowAction): 

35 name = "delete_user" 

36 resource_type = "users" 

37 

38 async def execute_audited(self, record, ctx): 

39 ... 

40 

41 Attributes: 

42 resource_type: The resource category recorded in the audit entry. 

43 Must be set by each subclass. 

44 """ 

45 

46 resource_type: str = "" 

47 

48 @abstractmethod 

49 async def execute_audited( 

50 self, record_or_records: _RT, ctx: ActionContext 

51 ) -> Result[_OT, ActionError]: ... 

52 

53 def capture_before(self, record: _RT) -> dict[str, Any] | None: 

54 return None 

55 

56 def capture_after(self, record: _RT, outcome: _OT) -> dict[str, Any] | None: 

57 return None 

58 

59 def resource_id_of(self, record: _RT) -> str: 

60 if record is None: 

61 return "" 

62 if isinstance(record, dict): 

63 return str(record.get("id", "")) 

64 return str(getattr(record, "id", record)) 

65 

66 async def execute( 

67 self, record_or_records: _RT, ctx: ActionContext 

68 ) -> Result[_OT, ActionError]: 

69 """Sealed execute() — wraps execute_audited() with audit writes.""" 

70 before_snapshot = self.capture_before(record_or_records) 

71 

72 result = await self.execute_audited(record_or_records, ctx) 

73 

74 outcome_value: AuditOutcome 

75 after_snapshot: dict[str, Any] = {} 

76 

77 if result.is_ok(): 

78 outcome_value = AuditOutcome.SUCCESS 

79 raw_after = self.capture_after(record_or_records, result.unwrap()) 

80 if raw_after is not None: 

81 after_snapshot = raw_after 

82 else: 

83 outcome_value = AuditOutcome.ERRORED 

84 

85 admin_user_id = "" 

86 if ctx.user is not None: 

87 admin_user_id = str( 

88 getattr(ctx.user, "user_id", None) 

89 or getattr(ctx.user, "id", None) 

90 or "" 

91 ) 

92 

93 resource_id = ctx.record_id or self.resource_id_of(record_or_records) 

94 

95 entry = AuditEntry( 

96 admin_user_id=admin_user_id, 

97 action=self.name, 

98 resource_type=self.resource_type or ctx.resource_name, 

99 resource_id=resource_id or None, 

100 outcome=outcome_value, 

101 before=before_snapshot or {}, 

102 after=after_snapshot, 

103 correlation_id=ctx.correlation_id, 

104 request_id=ctx.request_id, 

105 request_ip=ctx.request_ip, 

106 metadata=dict(ctx.metadata), 

107 ) 

108 

109 writer = ctx.audit_writer 

110 if writer is None: 

111 logger.warning( 

112 "admin.audited_action_no_writer", 

113 action=self.name, 

114 resource_type=self.resource_type, 

115 ) 

116 else: 

117 await writer.write(entry) 

118 

119 return result 

120 

121 

122__all__ = ["AuditedAction"]