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

1"""Admin resilience utilities. 

2 

3Provides the transaction() context manager for CRUD operations (FWK-11) 

4and AuditRepositoryMixin (FWK-12). 

5 

6Configuration types for circuit breaker, retry, and timeout policies live in 

7``lexigram.contracts.infra.resilience.models`` — import them from there. 

8""" 

9 

10from __future__ import annotations 

11 

12from collections.abc import AsyncGenerator, Callable 

13from contextlib import asynccontextmanager 

14from dataclasses import dataclass, field 

15from datetime import UTC, datetime 

16from typing import TYPE_CHECKING, Any, Generic, TypeVar 

17 

18from lexigram.contracts.admin.audit_logger import AdminAuditLoggerProtocol 

19 

20if TYPE_CHECKING: 

21 from lexigram.contracts.data import UnitOfWorkProtocol 

22 

23T = TypeVar("T") 

24F = TypeVar("F", bound=Callable[..., Any]) 

25 

26 

27# ============================================================================ 

28# Transaction Context Manager 

29# ============================================================================ 

30 

31 

32@asynccontextmanager 

33async def transaction( 

34 uow: UnitOfWorkProtocol, 

35) -> AsyncGenerator[UnitOfWorkProtocol, None]: 

36 """Async context manager that wraps a :class:`UnitOfWorkProtocol` boundary. 

37 

38 The ``UnitOfWorkProtocol`` is responsible for beginning the transaction on 

39 ``__aenter__`` and rolling back on any unhandled exception in 

40 ``__aexit__``. This helper simply enters the UoW context and yields it so 

41 callers can work with repositories, then exits cleanly on success (commit) 

42 or failure (rollback) via the UoW's own ``__aexit__`` semantics. 

43 

44 The ``uow`` **must** be injected via the DI container — never created ad 

45 hoc. Callers obtain it via constructor injection from 

46 ``DatabaseProviderProtocol.get_unit_of_work()``. 

47 

48 Args: 

49 uow: A :class:`~lexigram.contracts.data.UnitOfWorkProtocol` instance. 

50 

51 Yields: 

52 The same ``uow`` instance so callers can access repositories. 

53 

54 Example:: 

55 

56 class UserService: 

57 def __init__(self, db: DatabaseProviderProtocol) -> None: 

58 self._db = db 

59 

60 async def create(self, data: dict) -> User: 

61 async with transaction(self._db.get_unit_of_work()) as uow: 

62 user = User(**data) 

63 uow.register_new(user) 

64 await uow.commit() 

65 return user 

66 """ 

67 async with uow: 

68 yield uow 

69 

70 

71# ============================================================================ 

72# Audit Mixin 

73# ============================================================================ 

74 

75 

76@dataclass 

77class AuditEntry: 

78 """Audit log entry.""" 

79 

80 action: str 

81 resource_type: str 

82 resource_id: Any 

83 user_id: Any 

84 timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) 

85 changes: dict[str, Any] | None = None 

86 metadata: dict[str, Any] | None = None 

87 

88 

89class InMemoryAuditLogger: 

90 """In-memory audit logger for development/testing.""" 

91 

92 def __init__(self) -> None: 

93 self._entries: list[AuditEntry] = [] 

94 

95 async def log( 

96 self, 

97 action: str, 

98 resource_type: str, 

99 resource_id: Any, 

100 user_id: Any, 

101 changes: dict[str, Any] | None = None, 

102 metadata: dict[str, Any] | None = None, 

103 ) -> None: 

104 """Log an audit entry.""" 

105 entry = AuditEntry( 

106 action=action, 

107 resource_type=resource_type, 

108 resource_id=resource_id, 

109 user_id=user_id, 

110 changes=changes, 

111 metadata=metadata, 

112 ) 

113 self._entries.append(entry) 

114 

115 def get_entries( 

116 self, 

117 resource_type: str | None = None, 

118 resource_id: Any | None = None, 

119 action: str | None = None, 

120 ) -> list[AuditEntry]: 

121 """Get audit entries with optional filtering.""" 

122 entries = self._entries 

123 

124 if resource_type: 

125 entries = list(filter(lambda e: e.resource_type == resource_type, entries)) 

126 if resource_id: 

127 entries = list(filter(lambda e: e.resource_id == resource_id, entries)) 

128 if action: 

129 entries = list(filter(lambda e: e.action == action, entries)) 

130 

131 return entries 

132 

133 

134class AuditRepositoryMixin(Generic[T]): 

135 """Mixin adding audit logging to repository operations. 

136 

137 Add to your data source class to automatically log CRUD operations. 

138 

139 Example: 

140 >>> class UserDataSource(AuditRepositoryMixin[User]): 

141 ... resource_type = "users" 

142 ... 

143 ... async def create(self, data: dict, user: Any) -> User: 

144 ... result = await super().create(data, user) 

145 ... await self._audit_create(result, user) 

146 ... return result 

147 """ 

148 

149 resource_type: str = "unknown" 

150 _audit_logger: AdminAuditLoggerProtocol | None = None 

151 

152 def set_audit_logger(self, logger: AdminAuditLoggerProtocol) -> None: 

153 """Set the audit logger.""" 

154 self._audit_logger = logger 

155 

156 async def _audit_create(self, record: T, user: Any) -> None: 

157 """Log a create operation.""" 

158 if self._audit_logger: 

159 record_id = getattr(record, "id", None) 

160 user_id = getattr(user, "id", None) if user else None 

161 await self._audit_logger.log( 

162 action="create", 

163 resource_type=self.resource_type, 

164 resource_id=record_id, 

165 user_id=user_id, 

166 changes={"created": True}, 

167 ) 

168 

169 async def _audit_update( 

170 self, 

171 record: T, 

172 changes: dict[str, Any], 

173 user: Any, 

174 ) -> None: 

175 """Log an update operation.""" 

176 if self._audit_logger: 

177 record_id = getattr(record, "id", None) 

178 user_id = getattr(user, "id", None) if user else None 

179 await self._audit_logger.log( 

180 action="update", 

181 resource_type=self.resource_type, 

182 resource_id=record_id, 

183 user_id=user_id, 

184 changes=changes, 

185 ) 

186 

187 async def _audit_delete(self, record_id: Any, user: Any) -> None: 

188 """Log a delete operation.""" 

189 if self._audit_logger: 

190 user_id = getattr(user, "id", None) if user else None 

191 await self._audit_logger.log( 

192 action="delete", 

193 resource_type=self.resource_type, 

194 resource_id=record_id, 

195 user_id=user_id, 

196 ) 

197 

198 async def _audit_bulk_action( 

199 self, 

200 action: str, 

201 record_ids: list[Any], 

202 user: Any, 

203 metadata: dict[str, Any] | None = None, 

204 ) -> None: 

205 """Log a bulk operation.""" 

206 if self._audit_logger: 

207 user_id = getattr(user, "id", None) if user else None 

208 await self._audit_logger.log( 

209 action=f"bulk_{action}", 

210 resource_type=self.resource_type, 

211 resource_id=record_ids, 

212 user_id=user_id, 

213 metadata=metadata, 

214 ) 

215 

216 

217__all__ = [ 

218 "AdminAuditLoggerProtocol", 

219 "AuditEntry", 

220 # Audit 

221 "AuditRepositoryMixin", 

222 # Transaction context manager 

223 "InMemoryAuditLogger", 

224 "transaction", 

225]