from bson import DBRef
from .abstract import BaseDataObject, I18nErrorDict
from .i18n import N_
__all__ = ('List', 'Dict', 'Reference')
[docs]class List(BaseDataObject, list):
__slots__ = ('inner_field', '_modified')
def __init__(self, inner_field, *args, **kwargs):
super().__init__(*args, **kwargs)
self._modified = False
self.inner_field = inner_field
def __setitem__(self, key, obj):
obj = self.inner_field.deserialize(obj)
super().__setitem__(key, obj)
self.set_modified()
def __delitem__(self, key):
super().__delitem__(key)
self.set_modified()
[docs] def append(self, obj):
obj = self.inner_field.deserialize(obj)
ret = super().append(obj)
self.set_modified()
return ret
[docs] def insert(self, i, obj):
obj = self.inner_field.deserialize(obj)
ret = super().insert(i, obj)
self.set_modified()
return ret
[docs] def pop(self, *args, **kwargs):
ret = super().pop(*args, **kwargs)
self.set_modified()
return ret
[docs] def clear(self, *args, **kwargs):
ret = super().clear(*args, **kwargs)
self.set_modified()
return ret
[docs] def remove(self, *args, **kwargs):
ret = super().remove(*args, **kwargs)
self.set_modified()
return ret
[docs] def reverse(self, *args, **kwargs):
ret = super().reverse(*args, **kwargs)
self.set_modified()
return ret
[docs] def sort(self, *args, **kwargs):
ret = super().sort(*args, **kwargs)
self.set_modified()
return ret
[docs] def extend(self, iterable):
iterable = [self.inner_field.deserialize(obj) for obj in iterable]
ret = super().extend(iterable)
self.set_modified()
return ret
def __repr__(self):
return '<object %s.%s(%s)>' % (
self.__module__, self.__class__.__name__, list(self))
[docs] def is_modified(self):
if self._modified:
return True
if self and isinstance(self[0], BaseDataObject):
return any(obj.is_modified() for obj in self)
return False
[docs] def set_modified(self):
self._modified = True
[docs] def clear_modified(self):
self._modified = False
if self and isinstance(self[0], BaseDataObject):
for obj in self:
obj.clear_modified()
[docs]class Dict(BaseDataObject, dict):
__slots__ = ('key_field', 'value_field', '_modified')
def __init__(self, key_field, value_field, *args, **kwargs):
super().__init__(*args, **kwargs)
self._modified = False
self.key_field = key_field
self.value_field = value_field
def __setitem__(self, key, obj):
key = self.key_field.deserialize(key) if self.key_field else key
obj = self.value_field.deserialize(obj) if self.value_field else obj
super().__setitem__(key, obj)
self.set_modified()
def __delitem__(self, key):
super().__delitem__(key)
self.set_modified()
[docs] def pop(self, *args, **kwargs):
ret = super().pop(*args, **kwargs)
self.set_modified()
return ret
[docs] def popitem(self, *args, **kwargs):
ret = super().popitem(*args, **kwargs)
self.set_modified()
return ret
[docs] def setdefault(self, key, obj=None):
key = self.key_field.deserialize(key) if self.key_field else key
obj = self.value_field.deserialize(obj) if self.value_field else obj
ret = super().setdefault(key, obj)
self.set_modified()
return ret
[docs] def update(self, other):
new = {
self.key_field.deserialize(k) if self.key_field else k:
self.value_field.deserialize(v) if self.value_field else v
for k, v in other.items()
}
super().update(new)
self.set_modified()
def __repr__(self):
return '<object %s.%s(%s)>' % (
self.__module__, self.__class__.__name__, dict(self))
[docs] def is_modified(self):
if self._modified:
return True
if self and any(isinstance(v, BaseDataObject) for v in self.values()):
return any(obj.is_modified() for obj in self.values())
return False
[docs] def set_modified(self):
self._modified = True
[docs] def clear_modified(self):
self._modified = False
if self and any(isinstance(v, BaseDataObject) for v in self.values()):
for obj in self.values():
obj.clear_modified()
[docs]class Reference:
error_messages = I18nErrorDict(not_found=N_('Reference not found for document {document}.'))
def __init__(self, document_cls, pk):
self.document_cls = document_cls
self.pk = pk
self._document = None
[docs] def fetch(self, no_data=False, force_reload=False, projection=None):
"""
Retrieve from the database the referenced document
:param no_data: if True, the caller is only interested in whether
the document is present in database. This means the
implementation may not retrieve document's data to save bandwidth.
:param force_reload: if True, ignore any cached data and reload referenced
document from database.
:param projection: if supplied, this is a dictionary or list describing
a projection which limits the data returned from database.
"""
raise NotImplementedError
@property
def exists(self):
"""
Check if the reference document exists in the database.
"""
raise NotImplementedError
def __repr__(self):
return '<object %s.%s(document=%s, pk=%r)>' % (
self.__module__, self.__class__.__name__, self.document_cls.__name__, self.pk)
def __eq__(self, other):
if isinstance(other, self.document_cls):
return other.pk == self.pk
if isinstance(other, Reference):
return self.pk == other.pk and self.document_cls == other.document_cls
if isinstance(other, DBRef):
return self.pk == other.id and self.document_cls.collection.name == other.collection
return NotImplemented