Coverage for object_streams/sources/model.py: 92%

1"""Source adapter for direct model changes.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Iterable 

6from collections.abc import Sequence 

7from dataclasses import dataclass 

8from typing import Protocol 

9 

10from django.db import models 

11 

12from object_streams.events import ObjectRef 

13 

14 

15__all__ = ("ModelSource", "Source") 

16 

17 

18class Source(Protocol): 

19 facet: str 

20 

21 def subjects_for_source(self, instance: models.Model) -> Iterable[ObjectRef]: 

22 """Return stream subjects affected by a changed source instance.""" 

23 

24 def changed_fields(self, instance: models.Model) -> Sequence[str]: 

25 """Return changed subject fields when the source can report them.""" 

26 

27 

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

29class ModelSource: 

30 """Source adapter for events where the source row and subject row match.""" 

31 

32 model: type[models.Model] 

33 facet: str = "object" 

34 

35 def matches(self, instance: models.Model) -> bool: 

36 return isinstance(instance, self.model) 

37 

38 def subjects_for_source(self, instance: models.Model) -> Iterable[ObjectRef]: 

39 if not self.matches(instance): 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 return () 

41 return (ObjectRef.from_instance(instance),) 

42 

43 def changed_fields(self, instance: models.Model) -> Sequence[str]: 

44 return ()