1. ホーム
  2. パイソン

[解決済み】SqlAlchemyの結果をJSONにシリアライズする方法は?

2022-04-21 12:40:59

質問

Django は、DB から返された ORM モデルを JSON 形式に自動シリアライズするいくつかの良い機能を持っています。

SQLAlchemy のクエリ結果を JSON 形式にシリアライズするには?

試しに jsonpickle.encode が、クエリオブジェクト自体をエンコードしてしまいます。 試しに json.dumps(items) を返します。

TypeError: <Product('3', 'some name', 'some desc')> is not JSON serializable

SQLAlchemyのORMオブジェクトをJSON /XMLにシリアライズするのはそんなに難しいのでしょうか?デフォルトのシリアライザはないのでしょうか?最近、ORM のクエリ結果をシリアライズするのは、とても一般的なタスクです。

必要なのは、SQLAlchemy のクエリ結果を JSON または XML データとして返すことです。

SQLAlchemyオブジェクトのクエリ結果をJSON/XML形式で、javascriptデータグリッド(JQGrid http://www.trirand.com/blog/ )

解決方法は?

フラットな実装

このようなものを使うことができます。

from sqlalchemy.ext.declarative import DeclarativeMeta

class AlchemyEncoder(json.JSONEncoder):

    def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            # an SQLAlchemy class
            fields = {}
            for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                data = obj.__getattribute__(field)
                try:
                    json.dumps(data) # this will fail on non-encodable values, like other classes
                    fields[field] = data
                except TypeError:
                    fields[field] = None
            # a json-encodable dict
            return fields

        return json.JSONEncoder.default(self, obj)

を使ってJSONに変換します。

c = YourAlchemyClass()
print json.dumps(c, cls=AlchemyEncoder)

エンコードできないフィールドは無視されます('None'に設定してください)。

リレーションを自動展開しない(自己参照になり、永久にループする可能性があるため)。

再帰的で非循環的な実装

しかし、もし永遠にループしたいのであれば、次のような方法がある。

from sqlalchemy.ext.declarative import DeclarativeMeta

def new_alchemy_encoder():
    _visited_objs = []

    class AlchemyEncoder(json.JSONEncoder):
        def default(self, obj):
            if isinstance(obj.__class__, DeclarativeMeta):
                # don't re-visit self
                if obj in _visited_objs:
                    return None
                _visited_objs.append(obj)

                # an SQLAlchemy class
                fields = {}
                for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                    fields[field] = obj.__getattribute__(field)
                # a json-encodable dict
                return fields

            return json.JSONEncoder.default(self, obj)

    return AlchemyEncoder

そして、オブジェクトを使用してエンコードします。

print json.dumps(e, cls=new_alchemy_encoder(), check_circular=False)

これは、すべての子、その子、そのまた子...をエンコードすることになる。基本的には、データベース全体をエンコードする可能性があります。前にエンコードしたものに到達すると、それを「なし」としてエンコードします。

再帰的、循環的、選択的な実装です。

もう一つの選択肢は、おそらくより良い方法ですが、拡張したいフィールドを指定できるようにすることです。

def new_alchemy_encoder(revisit_self = False, fields_to_expand = []):
    _visited_objs = []

    class AlchemyEncoder(json.JSONEncoder):
        def default(self, obj):
            if isinstance(obj.__class__, DeclarativeMeta):
                # don't re-visit self
                if revisit_self:
                    if obj in _visited_objs:
                        return None
                    _visited_objs.append(obj)

                # go through each field in this SQLalchemy class
                fields = {}
                for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                    val = obj.__getattribute__(field)

                    # is this field another SQLalchemy object, or a list of SQLalchemy objects?
                    if isinstance(val.__class__, DeclarativeMeta) or (isinstance(val, list) and len(val) > 0 and isinstance(val[0].__class__, DeclarativeMeta)):
                        # unless we're expanding this field, stop here
                        if field not in fields_to_expand:
                            # not expanding this field: set it to None and continue
                            fields[field] = None
                            continue

                    fields[field] = val
                # a json-encodable dict
                return fields

            return json.JSONEncoder.default(self, obj)

    return AlchemyEncoder

で呼び出せるようになりました。

print json.dumps(e, cls=new_alchemy_encoder(False, ['parents']), check_circular=False)

SQLAlchemy の 'parents' というフィールドだけを展開するためなど。