import types
import sheraf.attributes
from ..types import SmallDict
[docs]class BaseModel(metaclass=BaseModelMetaclass):
"""
:class:`~sheraf.models.base.BaseModel` is the base class for every other model classes.
This is where the attribute reading and writing are handled.
Models can be used as dictionaries:
>>> class Cowboy(sheraf.Model):
... table = "cowboy"
... name = sheraf.SimpleAttribute()
...
>>> with sheraf.connection(): # doctest: +SKIP
... dict(Cowboy.create(name="George Abitbol"))
{'name': 'George Abitbol', '_creation': ...}
"""
attributes = {}
mapping = None
default_mapping = SmallDict
deleted = None
def __init__(self):
self.deleted = False
[docs] @classmethod
def create(cls, default=None, *args, **kwargs):
"""Create a model instance.
:param default: The data structure that will be used to store the instance state.
:param \\*\\*kwargs: Any model attribute can be initialized with the matching keyword.
:return: The newly created instance.
>>> class Cowboy(sheraf.Model):
... table = "cowboy"
... name = sheraf.SimpleAttribute(default="John Doe")
...
>>> with sheraf.connection(commit=True):
... cowboy = Cowboy.create()
... assert "John Doe" == cowboy.name
...
... cowboy = Cowboy.create(name="George Abitbol")
... assert "George Abitbol" == cowboy.name
...
... Cowboy.create(this_attribute_does_not_exist="something") # raises a TypeError
Traceback (most recent call last):
...
TypeError: TypeError: create() got an unexpected keyword argument 'this_attribute_does_not_exist'
The function can also create sub-instances recursively:
>>> class Horse(sheraf.InlineModel):
... name = sheraf.SimpleAttribute()
...
>>> class Cowboy(sheraf.Model):
... table = "cowboy"
... name = sheraf.SimpleAttribute(default="John Doe")
... horse = sheraf.InlineModelAttribute(Horse)
...
>>> with sheraf.connection():
... cowboy = Cowboy.create(name="George Abitbol", horse={"name": "Jolly Jumper"})
... cowboy.horse.name
'Jolly Jumper'
"""
instance = None
try:
mapping = (default or cls.default_mapping)()
instance = cls._decorate(mapping)
yield_callbacks = cls.call_callbacks(cls.cb_creation, instance)
instance.initialize(**kwargs)
cls.call_callbacks_again(yield_callbacks)
return instance
except Exception:
try:
if instance:
instance.delete()
except:
pass
raise
[docs] def delete(self):
"""Delete the current model instance.
>>> class MyModel(sheraf.Model):
... table = "my_model"
...
>>> with sheraf.connection():
... m = MyModel.create()
... assert m == MyModel.read(m.id)
... m_id = m.id
... m.delete()
... m.read(m_id)
Traceback (most recent call last):
...
sheraf.exceptions.ModelObjectNotFoundException: Id '...' not found in MyModel
"""
cls = self.__class__
model_yield_callbacks = cls.call_callbacks(cls.cb_deletion, self)
attributes_yield_callbacks = {}
for attribute_name, attribute in self.attributes.items():
attributes_yield_callbacks[attribute_name] = self.call_callbacks(
attribute.cb_deletion, self, old=getattr(self, attribute_name)
)
for attribute_name in self.attributes.keys():
self.delete_attribute(attribute_name)
self.deleted = True
for attribute_name in self.attributes.keys():
self.call_callbacks_again(attributes_yield_callbacks[attribute_name])
cls.call_callbacks_again(model_yield_callbacks)
def initialize(self, **kwargs):
invalid_attributes = [
attribute_name
for attribute_name in kwargs.keys()
if not self.attributes.get(attribute_name)
]
if invalid_attributes:
raise TypeError(
"TypeError: create() got an unexpected keywords arguments '{}'".format(
", ".join(invalid_attributes)
)
)
yield_callbacks = {}
additional_yield_callbacks = {}
additional_values = {}
additional_attributes = [
(attribute_name, attribute)
for attribute_name, attribute in self.attributes.items()
if (
not attribute.lazy
and attribute_name not in kwargs
and not attribute.is_created(self)
)
]
for attribute_name, value in kwargs.items():
attribute = self.attributes.get(attribute_name)
yield_callbacks[attribute_name] = self.call_callbacks(
attribute.cb_creation, self, new=value
)
for attribute_name, attribute in additional_attributes:
additional_values[attribute_name] = attribute.create(self)
additional_yield_callbacks = self.call_callbacks(
attribute.cb_creation, self, new=additional_values[attribute_name]
)
for attribute_name, value in kwargs.items():
self.set_attribute(attribute_name, value)
for attribute_name, attribute in additional_attributes:
self.set_attribute(attribute_name, additional_values[attribute_name])
for attribute_name in kwargs.keys():
self.call_callbacks_again(yield_callbacks[attribute_name])
for attribute_name, attribute in additional_attributes:
self.call_callbacks_again(additional_yield_callbacks)
@staticmethod
def call_callbacks(callbacks, *args, **kwargs):
yield_callbacks = []
for callback in callbacks:
res = callback(*args, **kwargs)
if isinstance(res, types.GeneratorType):
try:
next(res)
except StopIteration:
pass
yield_callbacks.append(res)
return yield_callbacks
@staticmethod
def call_callbacks_again(callbacks):
for callback in callbacks:
try:
next(callback)
except StopIteration:
pass
[docs] @classmethod
def on_creation(cls, *args, **kwargs):
"""
Decorator for callbacks to call on an instance creation. The callback
will be executed before the instance is created. If the callback yields,
the part after the yield will be executed after the creation.
The callback should take one attribute that is the model instance.
The callback can be freely named.
>>> class Cowboy(sheraf.Model):
... table = "cowboy_cb_creation"
... name = sheraf.StringAttribute()
...
>>>
... @Cowboy.on_creation
... def welcome_new_cowboys(cowboy):
... yield
... print(f"Welcome {cowboy.name}!")
...
>>> with sheraf.connection():
... george = Cowboy.create(name="George Abitbol")
Welcome George Abitbol!
"""
def wrapper(func):
cls.cb_creation.append(func)
return func
return wrapper if not args else wrapper(args[0])
[docs] @classmethod
def on_deletion(cls, *args, **kwargs):
"""
Decorator for callbacks to call on an instance deletion. The callback
will be executed before the instance is deleted. If the callback yields,
the part after the yield will be executed after the creation.
The callback should take one attribute that is the model instance.
The callback can be freely named.
>>> class Cowboy(sheraf.Model):
... table = "cowboy_cb_creation"
... name = sheraf.StringAttribute()
...
>>>
... @Cowboy.on_deletion
... def goodby_old_cowboys(cowboy):
... print(f"So long {cowboy.name}!")
...
>>> with sheraf.connection():
... george = Cowboy.create(name="George Abitbol")
... george.delete()
So long George Abitbol!
"""
def wrapper(func):
cls.cb_deletion.append(func)
return func
return wrapper if not args else wrapper(args[0])
@classmethod
def _decorate(cls, mapping):
instance = cls()
instance.mapping = mapping
return instance
@classmethod
def attribute_id(cls, name, attribute):
raise NotImplementedError
def __setattr__(self, name, value):
if name not in self.attributes:
super().__setattr__(name, value)
return
yield_callbacks = []
attribute = self.attributes.get(name)
if not attribute.is_created(self):
yield_callbacks = self.call_callbacks(
attribute.cb_creation, self, new=value
)
else:
yield_callbacks = self.call_callbacks(
attribute.cb_edition, self, new=value, old=getattr(self, name)
)
self.set_attribute(name, value)
self.call_callbacks_again(yield_callbacks)
def set_attribute(self, name, value):
value = self.attributes[name].write(self, value)
if self.attributes[name].write_memoization:
super().__setattr__(name, value)
def __delattr__(self, name):
if name not in self.attributes:
super().__delattr__(name)
return
attribute = self.attributes.get(name)
yield_callbacks = self.call_callbacks(
attribute.cb_deletion, self, old=getattr(self, name)
)
self.delete_attribute(name)
self.call_callbacks_again(yield_callbacks)
def delete_attribute(self, name):
self.attributes[name].delete(self)
try:
super().__delattr__(name)
except AttributeError:
return
def __getattribute__(self, name):
# TODO: Find a way to check that self.name exists or not without
# using try/except AttributeError.
# Because if a @property of this class, called `name` also raises
# an AttributeError, actually we cannot say from where the AttributeError
# was emitted.
try:
attribute = super().__getattribute__(name)
if not isinstance(attribute, sheraf.attributes.Attribute):
return attribute
except AttributeError as exc:
if f"object has no attribute '{name}'" not in str(exc):
raise
if self.deleted:
raise AttributeError("Cannot access attributes from a deleted object")
try:
attribute = self.attributes[name]
except KeyError:
raise AttributeError(name)
value = attribute.read(self)
if self.attributes[name].read_memoization:
super().__setattr__(name, value)
return value
[docs] def copy(self, **kwargs):
r"""
:param \*\*kwargs: Keywords arguments will be passed to
:func:`~sheraf.models.BaseModel.create` and thus
wont be copied.
:return: a copy of this instance.
"""
copy = self.__class__.create(**kwargs)
for name, attr in self.attributes.items():
if attr.is_created(self) and name not in kwargs:
setattr(copy, name, getattr(self, name))
return copy
[docs] def keys(self):
"""
:return: The instance attribute names.
"""
return self.attributes.keys()
def items(self):
return (
(key, self.attributes[key].read(self)) for key in self.attributes.keys()
)
[docs] def update(self, **kwargs):
"""Takes an arbitrary number of keywords arguments, and updates the
instance attributes matching the arguments.
This functions recursively calls :func:`sheraf.attributes.Attribute.edit` with `addition` and `edition` to `True`.
>>> class Horse(sheraf.InlineModel):
... name = sheraf.SimpleAttribute()
...
>>> class Cowboy(sheraf.Model):
... table = "people"
... name = sheraf.SimpleAttribute()
... horse = sheraf.InlineModelAttribute(Horse)
...
>>> with sheraf.connection(commit=True):
... george = Cowboy.create(name="George", horse={"name": "Centaurus"})
... george.update(name="*incognito*", horse={"name": "Jolly Jumper"})
... george.name
'*incognito*'
>>> with sheraf.connection():
... george.horse.name
'Jolly Jumper'
Note that sub-instances are also edited.
"""
self.edit(
kwargs, addition=True, edition=True, deletion=False, replacement=False
)
[docs] def assign(self, **kwargs):
"""Takes an arbitrary number of keywords arguments, and updates the
instance attributes matching the arguments.
This functions recursively calls :func:`sheraf.attributes.Attribute.edit` with `addition`, `edition` and `deletion` to `True`.
>>> class Arm(sheraf.InlineModel):
... name = sheraf.SimpleAttribute()
...
>>> class Cowboy(sheraf.Model):
... table = "people"
... name = sheraf.SimpleAttribute()
... arms = sheraf.SmallListAttribute(sheraf.InlineModelAttribute(Arm))
...
>>> with sheraf.connection(commit=True):
... george = Cowboy.create(name="Three arms cowboy", arms=[
... {"name": "Arm 1"}, {"name": "Arm 2"}, {"name": "Arm 3"},
... ])
... len(george.arms)
3
>>> with sheraf.connection(commit=True):
... len(george.assign(name="George Abitbol", arms=[
... {"name": "Superarm 1"}, {"name": "Superarm 2"},
... ]).arms)
2
>>> with sheraf.connection():
... george.arms[0].name
'Superarm 1'
George passed from 3 arms to only 2 because *assign* does remove sub instances. If we had called :func:`~sheraf.models.base.BaseModel.update` instead,
George would have his two first arms be renamed *superarms* but, the third one would not have been removed.
"""
return self.edit(
kwargs, addition=True, edition=True, deletion=True, replacement=False
)
[docs] def edit(
self,
value,
addition=True,
edition=True,
deletion=False,
replacement=False,
strict=False,
):
"""Take a dictionary and a set of options, and try to applies the
dictionary values to the instance structure.
:param value: The dictionary containing the values. The dictionary elements that
do not match the instance attributes will be ignored.
:param addition: If *True*, elements present in *value* and absent from the instance attributes will be added.
:param edition: If *True*, elements present in both *value* and the instance will be updated.
:param deletion: If *True*, elements present in the instance and absent from *value* will be deleted.
:param replacement: Like *edition*, but create a new element instead of updating one.
:param strict: If strict is *True*, every keys in value must be sheraf attributes of the current model. Default is *False*.
"""
invalid_attributes = [
attribute_name
for attribute_name in value.keys()
if attribute_name not in self.attributes
]
if strict and invalid_attributes:
raise TypeError(
"TypeError: edit() got unexpected keyword arguments '{}'".format(
", ".join(invalid_attributes)
)
)
yield_callbacks = {}
valid_attributes = [
attribute_name
for attribute_name in value.keys()
if attribute_name in self.attributes
]
for attribute_name in valid_attributes:
attribute = self.attributes[attribute_name]
new_value = value[attribute_name]
if not attribute.is_created(self):
yield_callbacks[attribute_name] = self.call_callbacks(
attribute.cb_creation, self, new=new_value
)
else:
old_value = attribute.read(self)
yield_callbacks[attribute_name] = self.call_callbacks(
attribute.cb_edition, self, new=new_value, old=old_value
)
for attribute_name in valid_attributes:
attribute = self.attributes[attribute_name]
new_value = value[attribute_name]
old_value = attribute.read(self)
updated = attribute.update(
old_value, new_value, addition, edition, deletion, replacement
)
self.set_attribute(attribute_name, updated)
for attribute_name in valid_attributes:
self.call_callbacks_again(yield_callbacks[attribute_name])
return self
def save(self):
for attr in self.attributes.values():
attr.save(self)
return self
[docs] def reset(self, attribute):
"""
Resets an attribute to its default value.
"""
self.__setattr__(attribute, self.attributes[attribute].create(self))
def __ne__(self, other):
return not self == other
def __eq__(self, other):
return (
hasattr(self, "mapping")
and hasattr(other, "mapping")
and self.mapping == other.mapping
)
def __getitem__(self, key):
return self.attributes[key].read(self)
def __setitem__(self, key, value):
value = self.attributes[key].write(self, value)
super().__setattr__(key, value)
def __contains__(self, key):
return key in self.attributes
def __repr__(self):
return f"<{self.__class__.__name__}>"