import itertools
import warnings
import sheraf.exceptions
from sheraf.models.base import BaseModel
from sheraf.models.base import BaseModelMetaclass
from sheraf.models.indexmanager import MultipleDatabaseIndexManager
from sheraf.models.indexmanager import SimpleIndexManager
[docs]class BaseIndexedModel(BaseModel, metaclass=BaseIndexedModelMetaclass):
"""
This class handles the whole indexation mechanism. The mechanisms
for reading or iterating over models in the database are handled
here.
"""
_primary_key = None
_is_first_instance = None
def __init__(self, *args, **kwargs):
self._identifier = None
self._raw_identifier = None
if not self.primary_key():
raise sheraf.exceptions.PrimaryKeyException(
"{} inherit from IndexedModel but has no primary key. Cannot create.".format(
self.__class__.__name__
)
)
super().__init__(*args, **kwargs)
@classmethod
def primary_key(cls):
if cls._primary_key is None:
for index_name, index in cls.indexes.items():
if not index.details.primary:
continue
if cls._primary_key is None:
cls._primary_key = index_name
else:
raise sheraf.exceptions.PrimaryKeyException(
"A model can have only one primary key. '{}' has '{}' and '{}'".format(
cls.__class__.__name__,
cls._primary_key,
index_name,
)
)
return cls._primary_key
[docs] @classmethod
def all(cls, index_name=None):
"""
:param index_name: Can be any index name. By default, this is the primary key.
:return: A :class:`~sheraf.queryset.QuerySet` containing all the
registered models in the given index.
"""
return sheraf.queryset.QuerySet(model_class=cls, primary_key=index_name)
def initialize(self, **kwargs):
for index_name, index_manager in self.indexes.items():
if not index_manager.details.auto:
continue
if (
self._is_indexable(index_manager.details, warn=True)
and not index_manager.table_initialized()
):
index_manager.table()
if self.primary_key() in kwargs:
identifier = kwargs[self.primary_key()]
del kwargs[self.primary_key()]
else:
identifier = self.attributes[self.primary_key()].create(self)
self.__setattr__(self.primary_key(), identifier)
super().initialize(**kwargs)
@classmethod
def _check_args(cls, *args, **kwargs):
if len(args) + len(kwargs) != 1:
raise TypeError(
"BaseIndexedModel.read (and variants) take only one positionnal or named parameter"
)
if args:
index_name = cls.primary_key()
key = args[0]
else:
index_name, key = list(kwargs.items())[0]
try:
index = cls.indexes[index_name]
except KeyError:
raise sheraf.exceptions.InvalidIndexException(
f"'{index_name}' is not a valid index"
)
return index, key
[docs] @classmethod
def read(cls, *args, **kwargs):
"""
Get a model instance from its identifier. If the model identifier is not valid, a
:class:`~sheraf.exceptions.ModelObjectNotFoundException` is raised.
The function takes only one parameter which key is the index where to
search, and which value is the index identifier. If the index is
multiple, a :class:`~sheraf.exceptions.MultipleIndexException` is
raised.
By default the index used is the `id` index.
:param *args*: The ``identifier`` of the model. There can be only one positionnal or
keyword argument.
:param *kwargs*: The ``identifier`` of the model. There can be only one positionnal or
keyword argument.
:return: The :class:`~sheraf.models.indexation.BaseIndexedModel` matching the id.
>>> class MyModel(sheraf.Model):
... table = "my_model"
... unique = sheraf.SimpleAttribute().index(unique=True)
... multiple = sheraf.SimpleAttribute().index()
...
>>> with sheraf.connection():
... m = MyModel.create(unique="A", multiple="B")
... assert MyModel.read(m.id) == m
... assert MyModel.read(unique="A") == m
...
>>> with sheraf.connection():
... MyModel.read("invalid")
Traceback (most recent call last):
...
ModelObjectNotFoundException
>>> with sheraf.connection():
... MyModel.read(multiple="B")
Traceback (most recent call last):
...
MultipleIndexException
"""
index, key = cls._check_args(*args, **kwargs)
if not index.details.unique:
raise sheraf.exceptions.MultipleIndexException(
"'{}' is a multiple index and cannot be used with 'read'".format(
index.details.key
)
)
return cls._decorate(cls._read_model_index(key, index))
[docs] @classmethod
def read_these(cls, *args, **kwargs):
"""
Get model instances from their identifiers. Unlike
:func:`~sheraf.models.indexation.BaseModel.read_these`,If an instance
identifiers does not exist, a :class:`~sheraf.exceptions.ModelObjectNotFoundException`
is raised.
The function takes only one parameter which key is the index where to
search, and which values are the index identifier.
By default the index used is the `id` index.
:return: A generator over the models matching the keys.
>>> class MyModel(sheraf.IntIndexedNamedAttributesModel):
... table = "my_model"
...
>>> with sheraf.connection():
... m1 = MyModel.create(id=1)
... m2 = MyModel.create(id=2)
...
... assert [m1, m2] == list(MyModel.read_these([m1.id, m2.id]))
... list(MyModel.read_these(["invalid"]))
Traceback (most recent call last):
...
sheraf.exceptions.ModelObjectNotFoundException: Id 'invalid' not found in MyModel, 'id' index
"""
index, keys = cls._check_args(*args, **kwargs)
if index.details.unique:
return (cls._decorate(cls._read_model_index(key, index)) for key in keys)
else:
return itertools.chain.from_iterable(
(
cls._decorate(mapping)
for mapping in cls._read_model_index(key, index)
)
for key in keys
)
[docs] @classmethod
def read_these_valid(cls, *args, **kwargs):
"""
Return model instances from an index. Unlike :func:`~sheraf.models.indexation.BaseModel.read_these`,
invalid index values are ignored.
The function takes only one parameter which key is the index where to
search, and which values are the index identifier.
By default the index used is the `id` index.
:return: A generator over the models matching the keys.
>>> class MyModel(sheraf.IntIndexedNamedAttributesModel):
... table = "my_model"
...
>>> with sheraf.connection():
... m1 = MyModel.create(id=1)
... m2 = MyModel.create(id=2)
...
... assert [m1, m2] == list(MyModel.read_these_valid([m1.id, m2.id]))
... assert [m1, m2] == list(MyModel.read_these_valid([m1.id, 42, m2.id]))
"""
index, keys = cls._check_args(*args, **kwargs)
if index.details.unique:
return (
cls._decorate(index.get_item(key))
for key in keys
if index.has_item(key)
)
else:
return itertools.chain.from_iterable(
(cls._decorate(mapping) for mapping in index.get_item(key))
for key in keys
if index.has_item(key)
)
@classmethod
def _read_model_index(cls, key, index):
try:
return index.get_item(key)
except KeyError:
raise sheraf.exceptions.ModelObjectNotFoundException(
cls, key, index.details.key
)
[docs] @classmethod
def index_table_rebuild(
cls, *args, callback=None, reset=True, start=None, end=None
):
"""
Resets a model indexation tables.
This method should be called if an attribute became indexed in an already
populated database.
:param *args: A list of index names to reset. If `None`, all the
indexes will be reseted. The primary index cannot be
resetted.
:param callback: A callback that is called each item a model instance is
iterated.
:param reset: If `True` the index tables are deleted before reindexaxtion.
Defaults to `True`.
"""
if reset:
cls.index_table_reset(*args)
if not args:
indexes = cls.indexes.values()
else:
indexes = [
index for index_name, index in cls.indexes.items() if index_name in args
]
for i, m in enumerate(cls.all()[start:end]):
for index in indexes:
if not index.details.primary:
index.add_item(m)
if callback and callback(i, m) is False:
break
@classmethod
def index_table_reset(cls, *args):
if not args:
indexes = cls.indexes.values()
else:
indexes = [
index for index_name, index in cls.indexes.items() if index_name in args
]
for index in indexes:
if not index.details.primary:
index.delete()
[docs] @classmethod
def filter(cls, predicate=None, **kwargs):
"""Shortcut for :func:`sheraf.queryset.QuerySet.filter`.
:return: :class:`sheraf.queryset.QuerySet`
"""
return sheraf.queryset.QuerySet(model_class=cls).filter(
predicate=predicate, **kwargs
)
[docs] @classmethod
def search(cls, *args, **kwargs):
"""
Shortcut for :func:`sheraf.queryset.QuerySet.search`.
:return: :class:`sheraf.queryset.QuerySet`
"""
return sheraf.queryset.QuerySet(model_class=cls).search(*args, **kwargs)
[docs] @classmethod
def order(cls, *args, **kwargs):
"""Shortcut for :func:`sheraf.queryset.QuerySet.order`.
:return: :class:`sheraf.queryset.QuerySet`
"""
return sheraf.queryset.QuerySet(model_class=cls).order(*args, **kwargs)
[docs] @classmethod
def get(cls, *args, **kwargs):
"""Shortcut for :func:`sheraf.queryset.QuerySet.filter` and
:func:`sheraf.queryset.QuerySet.get`. ``Cowboy.get(name="Peter")`` and
``Cowboy.filter(name="Peter").get()`` are equivalent.
:return: The instance of the model if the filter matches exactly one
instance. Otherwise, it raises a
:class:`~sheraf.exceptions.QuerySetUnpackException`.
>>> class Cowboy(sheraf.Model):
... table = "people"
... name = sheraf.SimpleAttribute()
... age = sheraf.SimpleAttribute()
...
>>> with sheraf.connection(commit=True):
... peter = Cowboy.create(name="Peter", age=30)
... steven = Cowboy.create(name="Steven", age=30)
... assert peter == Cowboy.get(name="Peter")
...
>>> with sheraf.connection():
... Cowboy.get()
Traceback (most recent call last):
...
sheraf.exceptions.TooManyValuesSetUnpackException: Trying to unpack a QuerySet with multiple elements <QuerySet model=Cowboy>
>>> with sheraf.connection():
... Cowboy.get(age=30)
Traceback (most recent call last):
...
sheraf.exceptions.TooManyValuesSetUnpackException: Trying to unpack a QuerySet with multiple elements <QuerySet model=Cowboy filters=(age=30)>
>>> with sheraf.connection():
... Cowboy.get(name="Unknown cowboy")
Traceback (most recent call last):
...
sheraf.exceptions.EmptyQuerySetUnpackException: Trying to unpack an empty QuerySet
"""
return cls.filter(*args, **kwargs).get()
@classmethod
def from_table(cls, table_name):
if table_name not in IndexedModelMetaclass.tables:
return None
return IndexedModelMetaclass.tables[table_name][1]
def __repr__(self):
try:
identifier = (
self.identifier
if self.mapping is not None and self.primary_key() in self.mapping
else None
)
except:
identifier = "???"
return "<{} {}={}>".format(
self.__class__.__name__, self.primary_key(), identifier
)
def __str__(self):
return str(self.identifier)
def __hash__(self):
return hash(self.identifier)
def __eq__(self, other):
return (
hasattr(self, self.primary_key())
and hasattr(other, self.primary_key())
and self.identifier == other.identifier
)
def set_attribute(self, name, value):
attribute = self.attributes.get(name)
should_update_index = (
attribute
and attribute.indexes
and (not attribute.is_created(self) or getattr(self, name) != value)
)
if should_update_index:
if (
attribute.is_created(self)
and attribute.has_primary_index
and getattr(self, name) != value
):
raise sheraf.SherafException(
f"Attribute '{name}' has a primary index and cannot be edited."
)
old_values = self.before_index_edition(attribute)
super().set_attribute(name, value)
if should_update_index:
self.after_index_edition(attribute, old_values)
def delete_attribute(self, name):
attribute = self.attributes.get(name)
old_values = self.before_index_edition(attribute)
super().delete_attribute(name)
self.after_index_edition(attribute, old_values, ignore_errors=True)
def before_index_edition(self, attribute):
old_index_values = {}
for index in attribute.indexes.values():
if not index.auto:
continue
if not self._is_indexable(index, warn=True):
continue
old_index_values[index] = index.get_model_index_keys(self)
return old_index_values
def after_index_edition(self, attribute, old_index_values, ignore_errors=True):
for index in attribute.indexes.values():
if not index.auto or not self._is_indexable(index):
continue
new_index_values = index.get_model_index_keys(self)
index_manager = self.indexes[index.key]
index_manager.update_item(
self,
old_index_values[index],
new_index_values,
ignore_errors=ignore_errors,
)
@property
def identifier(self):
"""
The identifier is the value of the primary_key for the current instance.
If the primary_key is 'id', then the identifier might be an UUID.
"""
if not self._identifier:
self._identifier = getattr(self, self.primary_key())
return self._identifier
@property
def raw_identifier(self):
if not self._identifier:
self._identifier = getattr(self, self.primary_key())
if not self._raw_identifier:
self._raw_identifier = self.mapping[self.primary_key()]
return self._raw_identifier
def _is_indexable(self, index, warn=False):
"""
To have its entries updated, an index must have its table previously
initialized, with the exception of the very first model instance in
the database.
"""
if self._is_first_instance is None:
self._is_first_instance = not self.index_manager().initialized()
index_manager = self.indexes[index.key]
index_table_exists = index_manager.table_initialized()
indexable = self._is_first_instance or index_table_exists
if not indexable and warn:
warnings.warn(
"New index in an already populated table. %s.%s will not be indexed. "
'Consider calling %s.index_table_rebuild("%s") to initialize the indexation table.'
% (
self.__class__.__name__,
index.key,
self.__class__.__name__,
index.key,
),
sheraf.exceptions.IndexationWarning,
stacklevel=5,
)
return indexable
[docs] def copy(self, **kwargs):
r"""
Copies a model.
The attributes carrying an unique index wont be copied, they will be
resetted instead.
:param \*\*kwargs: Keywords arguments will be passed to
:func:`~sheraf.models.BaseModel.create` and thus
wont be copied.
:return: a copy of this instance.
"""
unique_attributes = (
attribute
for index in self.indexes.values()
for attribute in index.details.attributes
if index.details.unique
)
for attribute in unique_attributes:
kwargs.setdefault(attribute.key(self), attribute.create(self))
return super().copy(**kwargs)
[docs] @classmethod
def count(cls, index_name=None):
"""
Counts the number of elements in an index.
:param index_name: The name of the index to count. By default
the primary index is used
"""
return cls.indexes[index_name or cls.primary_key()].count()
[docs] def index_keys(self, index_name):
"""
Returns the values generated for a given index.
This method is a helper to help debugging custom
:meth:`~sheraf.models.indexation.BaseIndexedModel.index_keys` methods.
:param index_name: The name of the index.
>>> class Horse(sheraf.Model):
... table = "index_keys_horse"
... name = sheraf.StringAttribute().index(
... index_keys_func=lambda name: name.lower()
... )
...
>>> with sheraf.connection():
... jolly = Horse.create(name="Jolly Jumper")
... jolly.index_keys("name")
{'jolly jumper'}
"""
return self.indexes[index_name].details.get_model_index_keys(self)
[docs] @classmethod
def search_keys(cls, **kwargs):
"""
Returns the values generated for a given index query.
This method is a helper to help debugging custom
:meth:`~sheraf.models.indexation.BaseIndexedModel.search_keys` methods.
:param index_name: The name of the index.
>>> class Horse(sheraf.Model):
... table = "search_keys_horse"
... name = sheraf.StringAttribute().index(
... index_keys_func=lambda name: name.lower()
... )
...
>>> Horse.search_keys(name="Jolly Jumper")
{'jolly jumper'}
"""
if len(kwargs) != 1:
raise AttributeError(
"search_keys takes exactly one argument, "
"and this must be a valid index"
)
index_name, value = list(kwargs.items())[0]
return cls.indexes[index_name].details.call_search_func(cls, value)
[docs] def is_indexed_with(self, **kwargs):
"""
Checks if a model would be returned if a given query would be done.
This method does not make an actual query, and thus should be faster
than making a real one.
:params **kwargs: A list of indexes to check.
>>> class Horse(sheraf.Model):
... table = "is_indexed_with_horse"
... name = sheraf.StringAttribute().index(
... index_keys_func=lambda name: name.lower()
... )
...
>>> with sheraf.connection():
... jolly = Horse.create(name="Jolly Jumper")
... jolly.is_indexed_with(name="Jolly JUMPER")
True
"""
return all(
any(
v in self.index_keys(index_name)
for v in self.search_keys(**{index_name: index_value})
)
for index_name, index_value in kwargs.items()
)
[docs]class IndexedModel(BaseIndexedModel, metaclass=IndexedModelMetaclass):
"""
:class:`~sheraf.models.indexation.IndexedModel` are the top-level
models in the database. They come with one or several indexes,
stored in a *table* at the root of the database. They must
have a **table** parameter defined and an **id** attribute.
They can have a **database_name** attribute. If it is set, then in a
default connection context:
- :func:`~sheraf.models.indexation.IndexedModel.create` will store the\
new model instances in this database;
- :func:`~sheraf.models.indexation.IndexedModel.read` and\
:func:`~sheraf.models.indexation.IndexedModel.all` (etc.) will read in\
priority in this database, and then in the default database.
- :func:`~sheraf.models.indexation.IndexedModel.delete` will try to delete\
the model from this database, and by default in the default database.
However, if a **database_name** is explicitly passed to
:func:`sheraf.databases.connection`, then every action will be
performed on this database, ignoring the model **database_name** attribute.
"""
database_name = None
table = None
index_manager_class = MultipleDatabaseIndexManager
id = sheraf.attributes.simples.SimpleAttribute().index(primary=True)
@classmethod
def index_manager(cls, index=None):
return cls.index_manager_class(
cls.database_name,
cls.table,
index,
)
def __init__(self, *args, **kwargs):
if "id" not in self.__class__.attributes:
raise sheraf.exceptions.PrimaryKeyException(
"{} inherit from IndexedModel but has no id attribute. Cannot create.".format(
self.__class__.__name__
)
)
super().__init__(*args, **kwargs)
[docs]class SimpleIndexedModel(BaseIndexedModel, metaclass=BaseIndexedModelMetaclass):
index_manager_class = SimpleIndexManager
@classmethod
def index_manager(cls, index=None):
return cls.index_manager_class(index)