Source code for bag.sqlalchemy.json_col

"""A transparent JSON column for SQLAlchemy models.

The content of this module was basically pasted from
and altered such that the value is never None.

from sqlalchemy.ext.mutable import Mutable
from sqlalchemy.types import TypeDecorator, Unicode
import json

[docs]class JSONEncodedDict(TypeDecorator): """Represents an immutable structure as a json-encoded string.""" impl = Unicode
[docs] def process_bind_param(self, value, dialect): if value is None: value = {} return json.dumps(value)
[docs] def process_result_value(self, value, dialect): if value is None: value = {} return json.loads(value)
[docs]class MutableDict(Mutable, dict): """A dict that knows when it gets changed. Usage in a SQLAlchemy model:: data = Column(MutableDict.as_mutable(JSONEncodedDict)) """
[docs] @classmethod def coerce(cls, key, value): """Convert plain dictionaries to MutableDict.""" if value is None: value = {} if not isinstance(value, MutableDict): if isinstance(value, dict): return MutableDict(value) # This call will raise ValueError return Mutable.coerce(key, value) else: return value
def __setitem__(self, key, value): """Detect dictionary set events and emit change events.""" dict.__setitem__(self, key, value) self.changed() def __delitem__(self, key): """Detect dictionary del events and emit change events.""" dict.__delitem__(self, key) self.changed()