import inspect
import sheraf
from sheraf.attributes import Attribute
from sheraf.attributes.collections import ListAttribute
from sheraf.attributes.collections import SetAttribute
from sheraf.models.indexation import BaseIndexedModel
[docs]class ModelLoader:
"""
Loads models from the base in a cache.
Inherited by most model types
(``Model[Attribute|List|Set|(Large)Dict]``)
"""
cache = {}
def __init__(self, model=None, **kwargs):
super().__init__(**kwargs)
self._model = model
modules = [inspect.getmodule(frame[0]) for frame in inspect.stack()]
self.module_paths = [module.__name__ for module in modules if module]
def load_model(self, modelpath):
if isinstance(modelpath, bytes):
modelpath = modelpath.decode("utf-8")
path = modelpath.split(".")
module_path, klass = ".".join(path[:-1]), path[-1]
module_paths = [module_path] if module_path else self.module_paths
for module_path in module_paths:
module = __import__(module_path, globals(), locals(), [klass], 0)
# pypy:
if module is None: # pragma: no cover
continue
try:
return getattr(module, klass)
except AttributeError:
pass
raise ImportError(f"Unable to load model '{klass}'")
def read(self, parent):
self.check_model(parent)
return super().read(parent)
def write(self, parent, value):
self.check_model(parent)
return super().write(parent, value)
@property
def model(self):
self.check_model()
return self._model
def check_model(self, parent=None):
if isinstance(self._model, (list, tuple)):
self._model = type(self._model)(
self._check_model(m, parent) for m in self._model
)
else:
self._model = self._check_model(self._model, parent)
def _check_model(self, model, parent):
if isinstance(model, (str, bytes)):
try:
return ModelLoader.cache[model]
except KeyError:
ModelLoader.cache[model] = self.load_model(model)
return ModelLoader.cache[model]
elif parent and not isinstance(model, type):
return type(
f"{parent.__class__.__name__}.{self.key(parent)}",
(model.__class__,),
model.attributes,
)
else:
return model
[docs]class AttributeLoader(ModelLoader):
def __init__(self, attribute=None, **kwargs):
self._attribute = attribute
super().__init__(**kwargs)
def check_attribute(self, attribute_name, parent):
if attribute_name not in self._model.attributes:
raise sheraf.SherafException(
f"'{attribute_name}' is not an attribute of {self._model.__name__}"
)
attribute = self._model.attributes[attribute_name]
if isinstance(attribute, (ListAttribute, SetAttribute)):
if not isinstance(attribute.attribute, ModelAttribute):
raise sheraf.SherafException(
f"'{self._model.__name__}.{attribute_name}' attribute should hold a 'ModelAttribute' or a 'IndexedModelAttribute' to be referenced by a 'ReverseModelAttribute'"
)
elif not isinstance(attribute, (ModelAttribute, IndexedModelAttribute)):
raise sheraf.SherafException(
f"'{self._model.__name__}.{attribute_name}' should be a 'ModelAttribute' or a 'IndexedModelAttribute' or a collection of these to be referenced by a 'ReversedModelAttribute'"
)
if attribute_name not in self.model.indexes:
raise sheraf.SherafException(
f"'{self._model.__name__}.{attribute_name}' should have an index to be referenced by a 'ReversedModelAttribute'"
)
return attribute
[docs]class ModelAttribute(ModelLoader, Attribute):
"""This attribute references another :class:`~sheraf.models.Model`.
:param model: The model type to store.
:type model: :class:`~sheraf.models.Model` or list of :class:`~sheraf.models.Model`
>>> class Horse(sheraf.Model):
... table = "horse"
... name = sheraf.SimpleAttribute()
...
>>> class Cowboy(sheraf.Model):
... table = "cowboy"
... name = sheraf.SimpleAttribute()
... mount = sheraf.ModelAttribute(Horse)
...
>>> with sheraf.connection(commit=True):
... jolly = Horse.create(name="Jolly Jumper")
... george = Cowboy.create(name="George Abitbol", mount=jolly)
...
... george.mount.name
'Jolly Jumper'
The referenced model can be dynamically created if its structure is passed through as a dict:
>>> with sheraf.connection(commit=True):
... peter = Cowboy.create(name="Peter", mount={"name": "Polly Pumper"})
... assert isinstance(peter.mount, Horse)
... peter.mount.name
'Polly Pumper'
When the referenced model is deleted, the value of the attribute becomes ``None``.
>>> with sheraf.connection(commit=True):
... george = Cowboy.read(george.id)
... jolly.delete()
... assert george.mount is None
Several model classes can be used, but this will be more memory consuming in the database.
>>> class Pony(sheraf.Model):
... table = "pony"
... name = sheraf.SimpleAttribute()
...
>>> class Cowboy(sheraf.Model):
... table = "cowboy"
... name = sheraf.SimpleAttribute()
... mount = sheraf.ModelAttribute((Horse, Pony))
...
>>> with sheraf.connection(commit=True):
... superpony = Pony.create(name="Superpony")
... peter = Cowboy.create(name="Peter", mount=superpony)
When several models are set, the first one is considered to be the default model.
The default model is used when there is a doubt on the read data, or in the case
of model creation with a dict.
"""
def __init__(self, model=None, **kwargs):
if not model:
raise sheraf.exceptions.SherafException(
"ModelAttribute requires model parameter."
)
kwargs["read_memoization"] = False
kwargs["write_memoization"] = False
super().__init__(default=None, model=model, **kwargs)
[docs] def index_keys(self, model):
"""
By default :class:`~sheraf.attributes.models.ModelAttribute` are indexed on
their identifier.
"""
if model is None:
return {None}
return {
(model.table, model.identifier)
if isinstance(self.model, (tuple, list))
else model.identifier
}
def deserialize(self, value):
if isinstance(value, tuple):
table, id_ = value
model = BaseIndexedModel.from_table(table)
if model is None:
self.check_model()
model = BaseIndexedModel.from_table(table)
else:
id_ = value
model = (
self.model[0] if isinstance(self.model, (list, tuple)) else self.model
)
if model is None:
self.check_model()
model = (
self.model[0]
if isinstance(self.model, (list, tuple))
else self.model
)
try:
return model.read(id_)
except (KeyError, sheraf.exceptions.ModelObjectNotFoundException):
return None
def serialize(self, value):
if value is None:
return None
elif isinstance(value, sheraf.IndexedModel):
return (
(value.table, value.identifier)
if isinstance(self.model, (tuple, list))
else value.identifier
)
elif self.model and isinstance(value, dict):
return self.model.create(**value).identifier
elif isinstance(value, tuple):
return value
else:
try:
return self.model.read(value).identifier
except sheraf.SherafException:
return None
[docs] def update(
self,
old_value,
new_value,
addition=True,
edition=True,
deletion=False,
replacement=False,
):
if replacement or old_value is None or not isinstance(new_value, dict):
return self.serialize(new_value)
return old_value.edit(new_value, addition, edition, deletion, replacement)
[docs]class ReverseModelAttribute(AttributeLoader, Attribute):
"""
Inverse reference to a :class:`~sheraf.attributes.models.ModelAttribute`.
:param model: The :class:`~sheraf.models.Model` to refer to.
:param attribute: The :class:`~sheraf.attributes.Attribute` in the model to refer to.
This model must be a :class:`~sheraf.attributes.models.ModelAttribute` or a
collection of :class:`~sheraf.attributes.models.ModelAttribute`.
The referenced attribute must be indexed.
>>> class Cowboy(sheraf.Model): # doctest: +SKIP
... table = "reverse_cowboys"
... name = sheraf.StringAttribute()
... horse = sheraf.ModelAttribute("Horse").index()
...
>>> class Horse(sheraf.Model): # doctest: +SKIP
... table = "reverse_horses"
... name = sheraf.StringAttribute()
... cowboy = sheraf.ReverseModelAttribute("Cowboy", "horse")
...
>>> with sheraf.connection(): # doctest: +SKIP
... george = Cowboy.create(name="George")
... horse = Horse.create(name="Jolly", cowboy=george)
... george.horse.name
"Jolly"
Collection attributes are also supported:
>>> class Cowboy(sheraf.Model): # doctest: +SKIP
... table = "reverse_multicowboys"
... name = sheraf.StringAttribute()
... horses = sheraf.LargeListAttribute(ModelAttribute("Horse").index())
...
>>> class Horse(sheraf.Model): # doctest: +SKIP
... table = "reverse_multihorses"
... name = sheraf.StringAttribute()
... cowboy = sheraf.ReverseModelAttribute("Cowboy", "horses")
...
>>> with sheraf.connection(): # doctest: +SKIP
... george = Cowboy.create(name="George")
... jolly = Horse.create(name="Jolly", cowboy=george)
... polly = Horse.create(name="Polly", cowboy=george)
... george.horses[0].name
... george.horses[1].name
"Jolly"
"Polly"
"""
def __init__(self, model, attribute, **kwargs):
kwargs["read_memoization"] = False
kwargs["write_memoization"] = False
super().__init__(default=None, model=model, attribute=attribute, **kwargs)
def read(self, parent):
self.check_model(parent)
self.check_attribute(self._attribute, parent)
search_args = {self._attribute: parent}
if not self.model.indexes[self._attribute].details.unique:
return self.model.search(**search_args)
try:
return self.model.search(**search_args).get()
except sheraf.QuerySetUnpackException:
return None
def write(self, parent, value):
self.check_model(parent)
self.check_attribute(self._attribute, parent)
self.delete(parent)
if value is None:
return None
if not isinstance(value, (list, set)):
value = [value]
# if values are ids, then we should load corresponding models
checked_values = []
for v in value:
if isinstance(v, self.model):
checked_values.append(v)
else:
try:
checked_values.append(self.model.read(v))
except sheraf.ModelObjectNotFoundException:
pass
for referent in checked_values:
attribute = self.model.attributes[self._attribute]
if isinstance(attribute, ModelAttribute):
setattr(referent, self._attribute, parent)
if isinstance(attribute, sheraf.ListAttribute):
setattr(
referent,
self._attribute,
getattr(referent, self._attribute) + [parent],
)
if isinstance(attribute, sheraf.SetAttribute):
setattr(
referent,
self._attribute,
getattr(referent, self._attribute) | {parent},
)
return checked_values
def delete(self, parent):
referents = self.read(parent)
if referents is None:
return
if isinstance(referents, self.model):
referents = [referents]
for referent in list(referents):
attribute = self.model.attributes[self._attribute]
if isinstance(attribute, ModelAttribute):
delattr(referent, self._attribute)
if isinstance(
attribute,
sheraf.ListAttribute,
):
new_values = [
e for e in getattr(referent, self._attribute) if e != parent
]
setattr(
referent,
self._attribute,
new_values,
)
if isinstance(
attribute,
sheraf.SetAttribute,
):
new_values = {
e for e in getattr(referent, self._attribute) if e != parent
}
setattr(
referent,
self._attribute,
new_values,
)
[docs]class InlineModelAttribute(ModelLoader, Attribute):
""":class:`~sheraf.attributes.models.ModelAttribute` behaves like a basic
model (i.e. have no indexation capability). The child attribute mapping is stored
in the parent mapping.
:param model: The model type to store.
:type model: :class:`~sheraf.models.inline.InlineModel`
>>> class Horse(sheraf.InlineModel):
... name = sheraf.StringAttribute()
...
>>> class Cowboy(sheraf.Model):
... table = "cowboy_inliner"
... name = sheraf.StringAttribute()
... horse = sheraf.InlineModelAttribute(Horse)
...
>>> with sheraf.connection(commit=True):
... jolly = Horse.create(name="Jolly Jumper")
... george = Cowboy.create(name="George", horse=jolly)
... george.horse.name
'Jolly Jumper'
"""
default_mapping = sheraf.types.SmallDict
def __init__(self, model=None, **kwargs):
kwargs.setdefault("default", self.default_mapping)
super().__init__(model=model, **kwargs)
def deserialize(self, value):
if value is None:
return None
return self.model._decorate(value)
def serialize(self, value):
if value is None:
return None
elif isinstance(value, sheraf.InlineModel):
return value.mapping
elif isinstance(value, dict):
return self.model.create(**value).mapping
else:
return self._default_value(value)
[docs] def update(
self,
old_value,
new_value,
addition=True,
edition=True,
deletion=False,
replacement=False,
):
if replacement or old_value is None:
return self.serialize(new_value)
return old_value.edit(new_value, addition, edition, deletion, replacement)
[docs]class IndexedModelAttribute(ModelLoader, Attribute):
"""
:class:`~sheraf.attributes.models.ModelAttribute` behaves like a classic model,
including the indexation capabilities. The child attribute mapping and all the
index mappings are is stored in the parent mapping.
:param model: The model type to store.
:type model: :class:`~sheraf.models.AttributeModel`
.. note:: The :class:`~sheraf.models.AttributeModel` must have a *primary* index.
>>> class Horse(sheraf.AttributeModel):
... name = sheraf.StringAttribute().index(primary=True)
... size = sheraf.IntegerAttribute().index()
...
>>> class Cowboy(sheraf.Model):
... table = "cowboy_indexer"
... name = sheraf.StringAttribute()
... horses = sheraf.IndexedModelAttribute(Horse)
...
>>> with sheraf.connection(commit=True):
... george = Cowboy.create(name="George Abitbol")
... jolly = george.horses.create(name="Jolly Jumper", size=32)
...
... assert jolly == george.horses.read("Jolly Jumper")
... assert jolly in george.horses.search(size=32)
"""
index_table_default = sheraf.types.SmallDict
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cb_registered = dict()
def decorate_model(self, parent):
for index in self.model.indexes.values():
key = self.key(parent)
if key not in parent.mapping:
parent.mapping[key] = self.index_table_default()
index.persistent = parent.mapping[key]
def read(self, parent):
if not self.cb_registered.get(parent):
self.cb_registered[parent] = True
@self.model.on_creation
@self.model.on_deletion
def update_parent_index(instance):
old_values = parent.before_index_edition(self)
yield
parent.after_index_edition(self, old_values)
self.decorate_model(parent)
return self.model
def write(self, parent, value):
self.decorate_model(parent)
if value is not None and not inspect.isclass(value):
for values_dict in value:
self.model.create(**values_dict)
return self.model
[docs] def index_keys(self, model):
return {instance.raw_identifier for instance in model.all()}
[docs] def search_keys(self, query):
return {query.raw_identifier if isinstance(query, self.model) else query}