refers to multiple classes of the same name within _decl_class_registry.
| 188 | |
| 189 | |
| 190 | class _MultipleClassMarker(_ClsRegistryToken): |
| 191 | """refers to multiple classes of the same name |
| 192 | within _decl_class_registry. |
| 193 | |
| 194 | """ |
| 195 | |
| 196 | __slots__ = "on_remove", "contents", "__weakref__" |
| 197 | |
| 198 | contents: Set[weakref.ref[Type[Any]]] |
| 199 | on_remove: CallableReference[Optional[Callable[[], None]]] |
| 200 | |
| 201 | def __init__( |
| 202 | self, |
| 203 | classes: Iterable[Type[Any]], |
| 204 | on_remove: Optional[Callable[[], None]] = None, |
| 205 | ): |
| 206 | self.on_remove = on_remove |
| 207 | self.contents = { |
| 208 | weakref.ref(item, self._remove_item) for item in classes |
| 209 | } |
| 210 | _registries.add(self) |
| 211 | |
| 212 | def remove_item(self, cls: Type[Any]) -> None: |
| 213 | self._remove_item(weakref.ref(cls)) |
| 214 | |
| 215 | def __iter__(self) -> Generator[Optional[Type[Any]], None, None]: |
| 216 | return (ref() for ref in self.contents) |
| 217 | |
| 218 | def attempt_get(self, path: List[str], key: str) -> Type[Any]: |
| 219 | if len(self.contents) > 1: |
| 220 | raise exc.InvalidRequestError( |
| 221 | 'Multiple classes found for path "%s" ' |
| 222 | "in the registry of this declarative " |
| 223 | "base. Please use a fully module-qualified path." |
| 224 | % (".".join(path + [key])) |
| 225 | ) |
| 226 | else: |
| 227 | ref = list(self.contents)[0] |
| 228 | cls = ref() |
| 229 | if cls is None: |
| 230 | raise NameError(key) |
| 231 | return cls |
| 232 | |
| 233 | def _remove_item(self, ref: weakref.ref[Type[Any]]) -> None: |
| 234 | self.contents.discard(ref) |
| 235 | if not self.contents: |
| 236 | _registries.discard(self) |
| 237 | if self.on_remove: |
| 238 | self.on_remove() |
| 239 | |
| 240 | def add_item(self, item: Type[Any]) -> None: |
| 241 | # protect against class registration race condition against |
| 242 | # asynchronous garbage collection calling _remove_item, |
| 243 | # [ticket:3208] and [ticket:10782] |
| 244 | modules = { |
| 245 | cls.__module__ |
| 246 | for cls in [ref() for ref in list(self.contents)] |
| 247 | if cls is not None |
no outgoing calls
no test coverage detected