Coverage for src/lumen/core/write_events.py: 89%

1""" 

2Transport-neutral collection of write events. 

3 

4The repository records what it wrote; *who* reacts to it is decided elsewhere. 

5This exists because write-time extensions (see ``lumen.cortex``) must run 

6**after** the writing transaction has committed — a transaction bundle commits 

7long after the individual ``create``/``update`` calls inside it returned. 

8 

9The repository therefore does not dispatch anything. It appends to whatever 

10collector is active on the current context, and the layer that owns the 

11transaction boundary (``ResourceService``, the bundle route) drains the 

12collector once the data is durable. 

13 

14Nothing is recorded unless a collector is active, so the normal path pays for 

15one ``ContextVar.get()`` per write. 

16""" 

17 

18from __future__ import annotations 

19 

20import contextlib 

21from collections.abc import Iterator 

22from contextvars import ContextVar 

23from dataclasses import dataclass 

24from datetime import datetime 

25from typing import Any, Literal 

26from uuid import UUID 

27 

28WriteOperation = Literal["create", "update", "delete"] 

29 

30 

31@dataclass(frozen=True) 

32class WriteEvent: 

33 """One durable change to one resource.""" 

34 

35 tenant_id: UUID 

36 operation: WriteOperation 

37 resource_type: str 

38 resource_id: str 

39 version_id: int 

40 last_updated: datetime | None 

41 #: The resource as stored. ``None`` for deletes. 

42 resource: dict[str, Any] | None 

43 

44 

45_collector: ContextVar[list[WriteEvent] | None] = ContextVar("lumen_write_events", default=None) 

46 

47 

48def record_write_event(event: WriteEvent) -> None: 

49 """Append ``event`` to the active collector, if any.""" 

50 sink = _collector.get() 

51 if sink is not None: 

52 sink.append(event) 

53 

54 

55@contextlib.contextmanager 

56def collect_write_events() -> Iterator[list[WriteEvent]]: 

57 """Collect the write events produced inside the block. 

58 

59 The list is filled as writes happen and is complete when the block exits. 

60 Nested collectors are independent: the inner block captures its own writes 

61 and the outer one does not see them, which keeps a plugin's own writes out 

62 of the batch that triggered it. 

63 """ 

64 events: list[WriteEvent] = [] 

65 token = _collector.set(events) 

66 try: 

67 yield events 

68 finally: 

69 _collector.reset(token) 

70 

71 

72@contextlib.contextmanager 

73def suspend_write_events() -> Iterator[None]: 

74 """Run a block with no active collector.""" 

75 token = _collector.set(None) 

76 try: 

77 yield 

78 finally: 

79 _collector.reset(token)