Simplify and document ModelStore
This commit is contained in:
parent
5f1044e96a
commit
6fa2d91b69
@ -1,60 +1,34 @@
|
|||||||
from typing import Dict, Iterator, MutableMapping, Set, Tuple, Type, Union
|
from collections import UserDict
|
||||||
|
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
|
from typing import Dict
|
||||||
|
|
||||||
from . import SyncId
|
from . import SyncId
|
||||||
from .model_item import ModelItem
|
|
||||||
from .model import Model
|
from .model import Model
|
||||||
|
|
||||||
KeyType = Union[Type[ModelItem], Tuple[Type, ...]]
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class ModelStore(MutableMapping):
|
class ModelStore(UserDict):
|
||||||
allowed_key_types: Set[KeyType] = field()
|
"""Dict of sync ID keys and `Model` values.
|
||||||
|
|
||||||
data: Dict[SyncId, Model] = field(init=False, default_factory=dict)
|
The dict keys must be the sync ID of `Model` values.
|
||||||
|
If a non-existent key is accessed, a corresponding `Model` will be
|
||||||
|
created, put into the interal `data` dict and returned.
|
||||||
|
"""
|
||||||
|
|
||||||
|
data: Dict[SyncId, Model] = field(default_factory=dict)
|
||||||
|
|
||||||
|
|
||||||
def __getitem__(self, key: SyncId) -> Model:
|
def __missing__(self, key: SyncId) -> Model:
|
||||||
try:
|
"""When accessing a non-existent model, create and return it."""
|
||||||
return self.data[key]
|
|
||||||
except KeyError:
|
|
||||||
if isinstance(key, tuple):
|
|
||||||
for i in key:
|
|
||||||
if not i:
|
|
||||||
raise ValueError(f"Empty string in key: {key!r}")
|
|
||||||
|
|
||||||
key_type = (key[0],) + \
|
model = Model(sync_id=key)
|
||||||
tuple(type(el) for el in key[1:])
|
|
||||||
else:
|
|
||||||
key_type = key # type: ignore
|
|
||||||
|
|
||||||
if key_type not in self.allowed_key_types:
|
|
||||||
raise TypeError(f"{key_type!r} not in allowed key types: "
|
|
||||||
f"{self.allowed_key_types!r}")
|
|
||||||
|
|
||||||
model = Model(key)
|
|
||||||
self.data[key] = model
|
self.data[key] = model
|
||||||
return model
|
return model
|
||||||
|
|
||||||
|
|
||||||
def __setitem__(self, key, item) -> None:
|
|
||||||
raise NotImplementedError()
|
|
||||||
|
|
||||||
|
|
||||||
def __delitem__(self, key: SyncId) -> None:
|
|
||||||
del self.data[key]
|
|
||||||
|
|
||||||
|
|
||||||
def __iter__(self) -> Iterator[SyncId]:
|
|
||||||
return iter(self.data)
|
|
||||||
|
|
||||||
|
|
||||||
def __len__(self) -> int:
|
|
||||||
return len(self.data)
|
|
||||||
|
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
|
"""Provide a nice overview of stored models when `print()` called."""
|
||||||
|
|
||||||
return "%s(\n %s\n)" % (
|
return "%s(\n %s\n)" % (
|
||||||
type(self).__name__,
|
type(self).__name__,
|
||||||
"\n ".join(sorted(str(v) for v in self.values())),
|
"\n ".join(sorted(str(v) for v in self.values())),
|
||||||
|
Loading…
x
Reference in New Issue
Block a user