Coverage for object_streams/registry.py: 77%

1"""Model registration for object stream subscriptions.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Iterable 

6from collections.abc import Iterator 

7from collections.abc import Mapping 

8from dataclasses import dataclass 

9from dataclasses import field 

10from typing import Any 

11 

12from django.apps import apps 

13from django.db import models 

14 

15from object_streams.exceptions import AlreadyRegistered 

16from object_streams.exceptions import NotRegistered 

17from object_streams.filters import FilterSetLike 

18from object_streams.filters import apply_filterset 

19from object_streams.visibility import AllowAllVisibilityPolicy 

20from object_streams.visibility import VisibilityPolicy 

21 

22 

23__all__ = ( 

24 "ObjectStreamRegistration", 

25 "ObjectStreamRegistry", 

26 "register", 

27 "registry", 

28) 

29 

30 

31def _normalize_model_label(model: type[models.Model] | str) -> str: 

32 if isinstance(model, str): 

33 return model 

34 return model._meta.label 

35 

36 

37def _resolve_model(model: type[models.Model] | str) -> type[models.Model]: 

38 if not isinstance(model, str): 38 ↛ 41line 38 didn't jump to line 41 because the condition on line 38 was always true

39 return model 

40 

41 try: 

42 app_label, model_name = model.split(".", 1) 

43 except ValueError as exc: 

44 msg = f"Model labels must use 'app_label.ModelName' format: {model!r}" 

45 raise LookupError(msg) from exc 

46 

47 model_class = apps.get_model(app_label, model_name) 

48 if model_class is None: 

49 msg = f"No installed model matches {model!r}." 

50 raise LookupError(msg) 

51 return model_class 

52 

53 

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

55class ObjectStreamRegistration: 

56 """Configuration used to subscribe to one Django model.""" 

57 

58 model: type[models.Model] 

59 filterset: type[FilterSetLike] | None = None 

60 visibility: VisibilityPolicy = field(default_factory=AllowAllVisibilityPolicy) 

61 sources: Iterable[Any] = field(default_factory=tuple) 

62 facets: Iterable[str] = field(default_factory=lambda: ("object",)) 

63 

64 def __post_init__(self): 

65 sources = tuple(self.sources) 

66 facets = {str(facet) for facet in self.facets} 

67 for source in sources: 

68 facet = getattr(source, "facet", None) 

69 if facet is not None: 69 ↛ 67line 69 didn't jump to line 67 because the condition on line 69 was always true

70 facets.add(str(facet)) 

71 

72 object.__setattr__(self, "sources", sources) 

73 object.__setattr__(self, "facets", frozenset(facets or {"object"})) 

74 

75 @property 

76 def model_label(self) -> str: 

77 return self.model._meta.label 

78 

79 def get_queryset( 

80 self, 

81 user: Any, 

82 filters: Mapping[str, Any] | None = None, 

83 *, 

84 action: str = "read", 

85 request: Any = None, 

86 ) -> models.QuerySet: 

87 queryset = self.visibility.get_queryset(user, self.model, action=action) 

88 return apply_filterset(self.filterset, queryset, filters, request=request) 

89 

90 

91class ObjectStreamRegistry: 

92 """In-memory registry for object stream model configuration.""" 

93 

94 def __init__(self): 

95 self._registrations: dict[str, ObjectStreamRegistration] = {} 

96 

97 def register( 

98 self, 

99 model: type[models.Model] | str | None = None, 

100 *, 

101 filterset: type[FilterSetLike] | None = None, 

102 visibility: VisibilityPolicy | None = None, 

103 sources: Iterable[Any] = (), 

104 facets: Iterable[str] | None = None, 

105 ): 

106 def do_register(model_class: type[models.Model]) -> ObjectStreamRegistration: 

107 registration = ObjectStreamRegistration( 

108 model=model_class, 

109 filterset=filterset, 

110 visibility=visibility or AllowAllVisibilityPolicy(), 

111 sources=tuple(sources), 

112 facets=tuple(facets or ("object",)), 

113 ) 

114 label = registration.model_label 

115 if label in self._registrations: 

116 msg = f"{label} is already registered for object streams." 

117 raise AlreadyRegistered(msg) 

118 self._registrations[label] = registration 

119 return registration 

120 

121 if model is None: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 return do_register 

123 return do_register(_resolve_model(model)) 

124 

125 def unregister(self, model: type[models.Model] | str) -> None: 

126 label = _normalize_model_label(model) 

127 try: 

128 del self._registrations[label] 

129 except KeyError as exc: 

130 msg = f"{label} is not registered for object streams." 

131 raise NotRegistered(msg) from exc 

132 

133 def get(self, model: type[models.Model] | str) -> ObjectStreamRegistration: 

134 label = _normalize_model_label(model) 

135 try: 

136 return self._registrations[label] 

137 except KeyError as exc: 

138 msg = f"{label} is not registered for object streams." 

139 raise NotRegistered(msg) from exc 

140 

141 def clear(self) -> None: 

142 self._registrations.clear() 

143 

144 def __contains__(self, model: type[models.Model] | str) -> bool: 

145 return _normalize_model_label(model) in self._registrations 

146 

147 def __iter__(self) -> Iterator[ObjectStreamRegistration]: 

148 return iter(self._registrations.values()) 

149 

150 

151registry = ObjectStreamRegistry() 

152register = registry.register