Coverage for object_streams/history/base.py: 100%

1"""Base protocols for history-backed source adapters.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import Any 

7from typing import Protocol 

8 

9from object_streams.events import SourceRef 

10 

11 

12__all__ = ("HistoryAdapter", "HistoryRecordRef") 

13 

14 

15@dataclass(frozen=True, slots=True) 

16class HistoryRecordRef: 

17 """Reference to an application history record that caused a stream event.""" 

18 

19 model: str 

20 pk: str 

21 history_model: str 

22 history_id: str 

23 

24 def as_source_ref(self) -> SourceRef: 

25 return SourceRef( 

26 model=self.model, 

27 pk=self.pk, 

28 history_model=self.history_model, 

29 history_id=self.history_id, 

30 ) 

31 

32 

33class HistoryAdapter(Protocol): 

34 def source_ref_for_history(self, history_record: Any) -> SourceRef: 

35 """Return a source reference for a history record."""