Coverage for src/lexigram/admin/controllers/resource/core.py: 73%

1"""Core state and lifecycle for the resource controller.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.admin.controllers.resource.meta import ResourceMeta, T 

8from lexigram.admin.data.data_source import IDataSource as DataSourceProtocol 

9from lexigram.logging import get_logger 

10 

11if TYPE_CHECKING: 

12 from lexigram.admin.controllers.resource import ResourceController 

13 

14logger = get_logger(__name__) 

15 

16 

17 

18class ResourceCoreMixin: 

19 """Shared state, audit, and revisions.""" 

20 

21 meta: ResourceMeta 

22 

23 # When True, DELETE calls soft-delete (sets deleted_at) instead of hard-delete. 

24 # Use RepositoryDataSource(soft_delete_enabled=True) to filter them in queries. 

25 soft_delete_enabled: bool = False 

26 def __init__(self: ResourceController, 

27 data_source: DataSourceProtocol[T] | None = None, 

28 meta: ResourceMeta | None = None, 

29 ): 

30 self._data_source = data_source 

31 if meta: 

32 self.meta = meta 

33 # Optional audit logger — set via set_audit_logger() or DI 

34 self._audit_logger: Any = None 

35 # Optional revision service — set via set_revision_service() or DI 

36 self._revision_service: Any = None 

37 

38 def set_audit_logger(self: ResourceController, audit_logger: Any) -> None: 

39 """Attach an audit logger for CRUD event tracking. 

40 

41 Args: 

42 audit_logger: Any object implementing ``async log(AuditEntry) -> None``. 

43 """ 

44 self._audit_logger = audit_logger 

45 

46 def set_revision_service(self: ResourceController, revision_service: Any) -> None: 

47 """Attach a :class:`~lexigram.admin.services.revisions.RevisionService`. 

48 

49 When set, a snapshot is recorded after every successful create or 

50 update. The service also exposes ``diff`` and ``revert_data`` for the 

51 revision history UI. 

52 

53 Args: 

54 revision_service: ``RevisionService`` instance. 

55 """ 

56 self._revision_service = revision_service 

57 async def _record_revision(self: ResourceController, 

58 request: Request, 

59 resource_id: str, 

60 data: dict[str, Any], 

61 comment: str = "", 

62 ) -> None: 

63 """Silently create a revision snapshot if a service is attached. 

64 

65 Args: 

66 request: Current HTTP request (actor identity extracted from state). 

67 resource_id: Record identifier. 

68 data: Full field snapshot to persist. 

69 comment: Optional human-readable description. 

70 """ 

71 if self._revision_service is None: 

72 return 

73 try: 

74 user = getattr(request.state, "user", None) 

75 actor_id = str(getattr(user, "id", getattr(user, "user_id", "system"))) 

76 resource_name = getattr(self.meta, "name", "resource") 

77 await self._revision_service.record( 

78 resource_name, 

79 resource_id, 

80 data, 

81 actor_id=actor_id, 

82 comment=comment, 

83 ) 

84 except Exception: # noqa: BLE001 

85 logger.warning( 

86 "Revision recording failed for resource %s/%s", 

87 getattr(self.meta, "name", ""), 

88 resource_id, 

89 ) 

90 async def _emit_audit(self: ResourceController, 

91 request: Request, 

92 action: str, 

93 item_id: str = "", 

94 old_values: dict[str, Any] | None = None, 

95 new_values: dict[str, Any] | None = None, 

96 outcome: str = "success", 

97 ) -> None: 

98 """Emit an audit log entry if a logger is attached. 

99 

100 Args: 

101 request: Current HTTP request (used for actor identity and metadata). 

102 action: Machine-readable action (e.g. ``"user.create"``). 

103 item_id: Identifier of the affected record. 

104 old_values: Field snapshot before the change. 

105 new_values: Field snapshot after the change. 

106 outcome: ``"success"`` or ``"failure"``. 

107 """ 

108 if self._audit_logger is None: 

109 return 

110 try: 

111 from lexigram.contracts.audit import AuditEntry 

112 

113 user = getattr(request.state, "user", None) 

114 actor_id = str(getattr(user, "id", getattr(user, "user_id", "anonymous"))) 

115 resource_name = getattr(self.meta, "name", "unknown") 

116 entry = AuditEntry( 

117 action=action, 

118 actor_id=actor_id, 

119 resource_type=resource_name, 

120 resource_id=str(item_id), 

121 outcome=outcome, 

122 old_values=old_values, 

123 new_values=new_values, 

124 metadata={ 

125 "ip": request.client.host if request.client else "", 

126 "user_agent": request.headers.get("user-agent", ""), 

127 "path": str(request.url), 

128 }, 

129 ) 

130 await self._audit_logger.log(entry) 

131 except Exception: # noqa: BLE001 

132 logger.warning( 

133 "Audit logging failed for action %s on %s/%s", 

134 action, 

135 getattr(self.meta, "name", ""), 

136 item_id, 

137 ) 

138 def get_data_source(self) -> DataSourceProtocol[T]: 

139 """Get the data source for this resource.""" 

140 if self._data_source is None: 

141 raise NotImplementedError("Subclass must implement get_data_source()") 

142 return self._data_source