File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/query.py", line 2399, in all
return list(self)
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/loading.py", line 86, in instances
util.raise_from_cause(err)
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/util/compat.py", line 199, in raise_from_cause
reraise(type(exception), exception, tb=exc_tb)
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/loading.py", line 71, in instances
rows = [proc(row) for row in fetch]
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/loading.py", line 432, in _instance
state.manager.dispatch.load(state, context)
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/event/attr.py", line 256, in __call__
fn(*args, **kw)
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/ext/mutable.py", line 451, in load
val = cls.coerce(key, val)
File "/srv/www/conferency/app/utils/customDataType.py", line 214, in coerce
return super(cls).coerce(key, value)
AttributeError: 'super' object has no attribute 'coerce'
class TrackedObject(object):
"""A base class for delegated change-tracking."""
_type_mapping = {}
def __init__(self, *args, **kwds):
self.logger = logging.getLogger(type(self).__name__)
self.parent = None
self.logger.debug('%s: __init__', self._repr())
super(TrackedObject, self).__init__(*args, **kwds)
def changed(self, message=None, *args):
"""Marks the object as changed.
If a `parent` attribute is set, the `changed()` method on the parent
will be called, propagating the change notification up the chain.
The message (if provided) will be debug logged.
"""
if message is not None:
self.logger.debug('%s: %s', self._repr(), message % args)
self.logger.debug('%s: changed', self._repr())
if self.parent is not None:
self.parent.changed()
elif isinstance(self, Mutable):
super(TrackedObject, self).changed()
@classmethod
def register(cls, origin_type):
"""Decorator for mutation tracker registration.
The provided `origin_type` is mapped to the decorated class such that
future calls to `convert()` will convert the object of `origin_type`
to an instance of the decorated class.
"""
def decorator(tracked_type):
"""Adds the decorated class to the `_type_mapping` dictionary."""
cls._type_mapping[origin_type] = tracked_type
return tracked_type
return decorator
@classmethod
def convert(cls, obj, parent):
"""Converts objects to registered tracked types
This checks the type of the given object against the registered tracked
types. When a match is found, the given object will be converted to the
tracked type, its parent set to the provided parent, and returned.
If its type does not occur in the registered types mapping, the object
is returned unchanged.
"""
replacement_type = cls._type_mapping.get(type(obj))
if replacement_type is not None:
new = replacement_type(obj)
new.parent = parent
return new
return obj
def convert_iterable(self, iterable):
"""Generator to `convert` every member of the given iterable."""
return (self.convert(item, self) for item in iterable)
def convert_items(self, items):
"""Generator like `convert_iterable`, but for 2-tuple iterators."""
return ((key, self.convert(value, self)) for key, value in items)
def convert_mapping(self, mapping):
"""Convenience method to track either a dict or a 2-tuple iterator."""
if isinstance(mapping, dict):
return self.convert_items(iteritems(mapping))
return self.convert_items(mapping)
def _repr(self):
"""Simple object representation."""
return '<%(namespace)s.%(type)s object at 0x%(address)0xd>' % {
'namespace': __name__,
'type': type(self).__name__,
'address': id(self)}
@TrackedObject.register(dict)
class TrackedDict(TrackedObject, dict):
"""A TrackedObject implementation of the basic dictionary."""
def __init__(self, source=(), **kwds):
super(TrackedDict, self).__init__(itertools.chain(
self.convert_mapping(source),
self.convert_mapping(kwds)))
def __setitem__(self, key, value):
self.changed('__setitem__: %r=%r', key, value)
super(TrackedDict, self).__setitem__(key, self.convert(value, self))
def __delitem__(self, key):
self.changed('__delitem__: %r', key)
super(TrackedDict, self).__delitem__(key)
def clear(self):
self.changed('clear')
super(TrackedDict, self).clear()
def pop(self, *key_and_default):
self.changed('pop: %r', key_and_default)
return super(TrackedDict, self).pop(*key_and_default)
def popitem(self):
self.changed('popitem')
return super(TrackedDict, self).popitem()
def update(self, source=(), **kwds):
self.changed('update(%r, %r)', source, kwds)
super(TrackedDict, self).update(itertools.chain(
self.convert_mapping(source),
self.convert_mapping(kwds)))
@TrackedObject.register(list)
class TrackedList(TrackedObject, list):
"""A TrackedObject implementation of the basic list."""
def __init__(self, iterable=()):
super(TrackedList, self).__init__(self.convert_iterable(iterable))
def __setitem__(self, key, value):
self.changed('__setitem__: %r=%r', key, value)
super(TrackedList, self).__setitem__(key, self.convert(value, self))
def __delitem__(self, key):
self.changed('__delitem__: %r', key)
super(TrackedList, self).__delitem__(key)
def append(self, item):
self.changed('append: %r', item)
super(TrackedList, self).append(self.convert(item, self))
def extend(self, iterable):
self.changed('extend: %r', iterable)
super(TrackedList, self).extend(self.convert_iterable(iterable))
def remove(self, value):
self.changed('remove: %r', value)
return super(TrackedList, self).remove(value)
def pop(self, index):
self.changed('pop: %d', index)
return super(TrackedList, self).pop(index)
def sort(self, cmp=None, key=None, reverse=False):
self.changed('sort')
super(TrackedList, self).sort(cmp=cmp, key=key, reverse=reverse)
class NestedMutableDict(TrackedDict, Mutable):
@classmethod
def coerce(cls, key, value):
if isinstance(value, cls):
return value
if isinstance(value, dict):
return cls(value)
return super(cls).coerce(key, value)
class NestedMutableList(TrackedList, Mutable):
@classmethod
def coerce(cls, key, value):
if isinstance(value, cls):
return value
if isinstance(value, list):
return cls(value)
return super(cls).coerce(key, value)
class NestedMutable(Mutable):
"""SQLAlchemy `mutable` extension with nested change tracking."""
@classmethod
def coerce(cls, key, value):
"""Convert plain dictionary to NestedMutable."""
if value is None:
return value
if isinstance(value, cls):
return value
if isinstance(value, dict):
return NestedMutableDict.coerce(key, value)
if isinstance(value, list):
return NestedMutableList.coerce(key, value)
return super(cls).coerce(key, value)
class MutableJson(JSONType):
"""JSON type for SQLAlchemy with change tracking at top level."""
class NestedMutableJson(JSONType):
"""JSON type for SQLAlchemy with nested change tracking."""
MutableDict.associate_with(MutableJson)
NestedMutable.associate_with(NestedMutableJson)
leon0707
commented
6 years ago