Coverage for src / lexigram / contracts / data / sql / unit_of_work.py: 0%

1"""Unit of Work protocol. 

2 

3The Unit of Work pattern tracks changes to entities and 

4coordinates their persistence in a single transaction. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any, Protocol, Self, runtime_checkable 

10 

11if TYPE_CHECKING: 

12 from lexigram.contracts.domain.events import DomainEvent 

13 

14 

15@runtime_checkable 

16class UnitOfWorkProtocol(Protocol): 

17 """Protocol for Unit of Work pattern. 

18 

19 Coordinates multiple repository operations within a single 

20 transaction boundary. 

21 

22 Example: 

23 ```python 

24 async with uow: 

25 user = await uow.users.get(user_id) 

26 user.email = new_email 

27 await uow.users.save(user) 

28 await uow.audit_log.save(AuditEntry(...)) 

29 await uow.commit() 

30 ``` 

31 """ 

32 

33 async def __aenter__(self) -> Self: 

34 """Begin unit of work scope.""" 

35 ... 

36 

37 async def __aexit__( 

38 self, 

39 exc_type: type[BaseException] | None, 

40 exc_val: BaseException | None, 

41 exc_tb: object, 

42 ) -> None: 

43 """End unit of work scope, rollback on exception.""" 

44 ... 

45 

46 async def commit(self) -> None: 

47 """Commit all changes in this unit of work.""" 

48 ... 

49 

50 async def rollback(self) -> None: 

51 """Rollback all changes in this unit of work.""" 

52 ... 

53 

54 def register_new(self, entity: Any) -> None: 

55 """Register a new entity for insertion.""" 

56 ... 

57 

58 # ------------------------------------------------------------------ 

59 # Event collection API 

60 # ------------------------------------------------------------------ 

61 

62 def register_event(self, event: DomainEvent) -> None: 

63 """Register a domain event with this unit of work. 

64 

65 Handlers or repositories may call this method when they detect that a 

66 domain event needs to be published once the transaction successfully 

67 commits. The unit of work is responsible for retaining the events until 

68 they are gathered by the application and dispatched via an event bus. 

69 """ 

70 ... 

71 

72 def collect_events(self) -> list[DomainEvent]: 

73 """Return and clear all domain events registered with this unit of 

74 work. 

75 

76 Calling this method yields the list of events that have been registered 

77 either explicitly via :meth:`register_event` or implicitly by the UoW 

78 when entities with ``collect_events`` semantics are registered. The 

79 returned list is cleared from the unit of work; subsequent calls will 

80 return an empty list until more events are registered. 

81 """ 

82 ... 

83 

84 def register_dirty(self, entity: Any) -> None: 

85 """Register an entity for update.""" 

86 ... 

87 

88 def register_deleted(self, entity: Any) -> None: 

89 """Register an entity for deletion.""" 

90 ... 

91 

92 

93__all__ = [ 

94 "UnitOfWorkProtocol", 

95]