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

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

2 

3from __future__ import annotations 

4 

5from typing import Any, Generic 

6 

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

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

9from starlette.requests import Request 

10 

11from lexigram.logging import get_logger 

12 

13logger = get_logger(__name__) 

14 

15 

16 

17class ResourceCoreMixin(Generic[T]): 

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

19 

20 meta: ResourceMeta 

21 

22 _data_source: DataSourceProtocol[T] | None 

23 _audit_logger: Any 

24 _revision_service: Any 

25 

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

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

28 soft_delete_enabled: bool = False 

29 def __init__( 

30 self, 

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

32 meta: ResourceMeta | None = None, 

33 ): 

34 self._data_source = data_source 

35 if meta: 

36 self.meta = meta 

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

38 self._audit_logger: Any = None 

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

40 self._revision_service: Any = None 

41 

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

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

44 

45 Args: 

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

47 """ 

48 self._audit_logger = audit_logger 

49 

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

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

52 

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

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

55 revision history UI. 

56 

57 Args: 

58 revision_service: ``RevisionService`` instance. 

59 """ 

60 self._revision_service = revision_service 

61 async def _record_revision(self, 

62 request: Request, 

63 resource_id: str, 

64 data: dict[str, Any], 

65 comment: str = "", 

66 ) -> None: 

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

68 

69 Args: 

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

71 resource_id: Record identifier. 

72 data: Full field snapshot to persist. 

73 comment: Optional human-readable description. 

74 """ 

75 if self._revision_service is None: 

76 return 

77 try: 

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

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

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

81 await self._revision_service.record( 

82 resource_name, 

83 resource_id, 

84 data, 

85 actor_id=actor_id, 

86 comment=comment, 

87 ) 

88 except Exception: # noqa: BLE001 

89 logger.warning( 

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

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

92 resource_id, 

93 ) 

94 async def _emit_audit(self, 

95 request: Request, 

96 action: str, 

97 item_id: str = "", 

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

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

100 outcome: str = "success", 

101 ) -> None: 

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

103 

104 Args: 

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

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

107 item_id: Identifier of the affected record. 

108 old_values: Field snapshot before the change. 

109 new_values: Field snapshot after the change. 

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

111 """ 

112 if self._audit_logger is None: 

113 return 

114 try: 

115 from lexigram.contracts.audit import AuditEntry 

116 

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

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

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

120 entry = AuditEntry( 

121 action=action, 

122 actor_id=actor_id, 

123 resource_type=resource_name, 

124 resource_id=str(item_id), 

125 outcome=outcome, 

126 old_values=old_values, 

127 new_values=new_values, 

128 metadata={ 

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

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

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

132 }, 

133 ) 

134 await self._audit_logger.log(entry) 

135 except Exception: # noqa: BLE001 

136 logger.warning( 

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

138 action, 

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

140 item_id, 

141 ) 

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

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

144 if self._data_source is None: 

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

146 return self._data_source