Coverage for src/lexigram/admin/audit/uow_writer.py: 0%

1"""UowAuditWriter — transactional audit writer participating in unit of work.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import Any, Protocol 

7 

8from lexigram.contracts.admin.audit_entry import AuditEntry 

9from lexigram.contracts.admin.audit_logger import AdminAuditLoggerProtocol 

10 

11 

12class UoWProtocol(Protocol): 

13 """Minimal unit-of-work interface for audit deferral.""" 

14 

15 def on_commit(self, callback: Callable[[], Any]) -> None: ... 

16 def on_rollback(self, callback: Callable[[], Any]) -> None: ... 

17 

18 

19class UowAuditWriter: 

20 """Audit writer that defers writes to the active unit of work. 

21 

22 If no active UoW is available (e.g. outside a request context), 

23 writes are dispatched directly to the underlying logger. 

24 """ 

25 

26 def __init__( 

27 self, 

28 logger: AdminAuditLoggerProtocol, 

29 uow_provider: Callable[[], UoWProtocol | None], 

30 ) -> None: 

31 self._logger = logger 

32 self._uow_provider = uow_provider 

33 

34 async def write(self, entry: AuditEntry) -> None: 

35 """Write *entry*, deferring to active UoW when available.""" 

36 uow = self._uow_provider() 

37 if uow is None: 

38 await self._logger.write(entry) 

39 return 

40 uow.on_commit(lambda: self._logger.write(entry)) 

41 

42 

43__all__ = ["UoWProtocol", "UowAuditWriter"]