json - 不跟踪更改的可变 JSON 对象类型 SQLAlchemy (sqlalchemy-json)

Question

好的:我正在使用 Flask-SQLAlchemy 和 PostgreSQL 构建一个 Flask 应用程序，我想在我的数据库中创建一个可变的 JSON 类型列。 SQLAlchemy 中有很多在线示例(以及这里有很多关于自定义可变对象类型的问题)。我找到了这个:sqlalchemy-json (完整的文章 by the author here )处理可变的 JSON 对象类型。从理论上讲，它提供了 JsonObject，一种 JSON 对象类型，在基本级别对 dict 和列表进行更改跟踪，以及 NestedJsonObject，一种 JSON 对象类型，对 dict 和列表进行嵌套更改跟踪。它看起来很甜蜜。

我一生都无法让它发挥作用，尽管我完全有信心。请注意:上述作者的文章没有显示在列中实现对象类型的示例，因此由于我是新手，我很可能将下一部分弄错了。不过我咨询了sqlalchemy.ext.mutable reference这似乎是对的。

这是我的models.py:

from application import app
from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin
import sqlalchemy_json # I'm using Alembic for migrations and make this import in my script.py.mako, too, in case that matters
from sqlalchemy_json import NestedJsonObject

db = SQLAlchemy(app)

class User(UserMixin, db.Model): #flask-sqlalchemy provides a Base declaration with db.Model
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), unique=True)
    data = db.Column(NestedJsonObject)

    def __init__(self, name, data):
        self.name = name
        self.data = data

    def __repr__(self):
        return "<User(id='%d', name='%s', connections='%r')>" % (self.id, self.name, self.data)

...

鉴于此，这是与我的用户合作的示例:

data = {}
data['a']= {'b': 'c', 'd': 'e'}
user = User( ... , data)
db.session.add(user)
db.session.commit()
#<User( ... data='{u'a': {u'b': u'c', u'd':u'e'}}')>

该初始提交有效。但随后的提交不会:

user = db.session.query(User).filter( ... ).first()
user.data['foo']={}
db.session.commit()
#<User( ... data='{u'a': {u'b': u'c', u'd':u'e'}}')>
user.data['foo']['bar'] = {'x': 'x', 'z': 'z'}
db.session.commit()
#<User( ... data='{u'a': {u'b': u'c', u'd':u'e'}}')>

你明白了。为清楚起见，以下是模块中带有作者注释的两个文件:

炼金术.py

请注意 .associate_with位于文件底部:

# Third-party modules
try:
  import simplejson as json
except ImportError:
  import json

import sqlalchemy
from sqlalchemy.ext import mutable

# Custom modules
from . import track


class NestedMutable(mutable.MutableDict, track.TrackedDict):
  """SQLAlchemy `mutable` extension dictionary with nested change tracking."""
  def __setitem__(self, key, value):
    """Ensure that items set are converted to change-tracking types."""
    super(NestedMutable, self).__setitem__(key, self.convert(value, self))

  @classmethod
  def coerce(cls, key, value):
    """Convert plain dictionary to NestedMutable."""
    if isinstance(value, cls):
      return value
    if isinstance(value, dict):
      return cls(value)
    return super(cls).coerce(key, value)


class _JsonTypeDecorator(sqlalchemy.TypeDecorator):
  """Enables JSON storage by encoding and decoding on the fly."""
  impl = sqlalchemy.String

  def process_bind_param(self, value, dialect):
    return json.dumps(value)

  def process_result_value(self, value, dialect):
    return json.loads(value)


class JsonObject(_JsonTypeDecorator):
  """JSON object type for SQLAlchemy with change tracking as base level."""


class NestedJsonObject(_JsonTypeDecorator):
  """JSON object type for SQLAlchemy with nested change tracking."""


mutable.MutableDict.associate_with(JsonObject)
NestedMutable.associate_with(NestedJsonObject)

跟踪文件

#!/usr/bin/python
"""This module contains the tracked object classes.
TrackedObject forms the basis for both the TrackedDict and the TrackedList.
A function for automatic conversion of dicts and lists to their tracked
counterparts is also included.
"""

# Standard modules
import itertools
import logging


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.logger.debug('%s: __init__', self._repr())
    self.parent = None
    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()

  @classmethod
  def register(cls, origin_type):
    """Registers the class decorated with this method as a mutation tracker.
    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.
    """
    obj_type = type(obj)
    for origin_type, replacement in cls._type_mapping.iteritems():
      if obj_type is origin_type:
        new = replacement(obj)
        new.parent = parent
        return new
    return obj

  @classmethod
  def convert_iterable(cls, iterable, parent):
    """Returns a generator that performs `convert` on every of its members."""
    return (cls.convert(item, parent) for item in iterable)

  @classmethod
  def convert_iteritems(cls, iteritems, parent):
    """Returns a generator like `convert_iterable` for 2-tuple iterators."""
    return ((key, cls.convert(value, parent)) for key, value in iteritems)

  @classmethod
  def convert_mapping(cls, mapping, parent):
    """Convenience method to track either a dict or a 2-tuple iterator."""
    if isinstance(mapping, dict):
      return cls.convert_iteritems(mapping.iteritems(), parent)
    return cls.convert_iteritems(mapping, parent)

  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),
        self.convert_mapping(kwds, self)))

  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),
        self.convert_mapping(kwds, self)))


@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, self))

  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, self))

  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)

还，

初始化文件

from .alchemy import NestedJsonObject, NestedMutable, JsonObject

__all__ = (
    'NestedJsonObject',
    'NestedMutable',
    'JsonObject'
)

感谢您阅读到这里。如果您有任何建议，请让我知道。如果这是重复的(我自己无法完全找到，但它们可能存在)，请将其标记为如此。此外，如果您自己发现上述代码对您的项目是一个很好的补充，请支持该模块的作者。我不是他/她。这是他/她的执照:

Copyright (c) 2014, Elmer de Looff <elmer.delooff@gmail.com>
All rights reserved.

Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.

2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

Answer 1

上网查了一下，sqlalchemy-json(感谢edelooff)是最好的选择，但是origin one只支持mutable dict，this fork由 torotil 更新解决了问题，感谢 torotil!

并将记录器放在 TrackedObject 中会导致深度复制问题，最好移到模块中。我 fork 了一个新的存储库来解决这个问题:Cysnake4713

另外，我使用 origin sqlalchemy.sql.sqltypes.JSON 而不是自定义的 JsonObject，看起来效果很好。

只需将 associate_with 部分替换为:

NestedMutable.associate_with(db.JSON)