Как сериализовать SqlAlchemy результат в JSON? - PullRequest
146 голосов
/ 17 февраля 2011

Django имеет несколько хороших автоматических сериализаций моделей 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.

Мне нужно просто вернуть JSON или XML-представление данных результата запроса SQLAlchemy.

Результат запроса объектов SQLAlchemy в формате JSON / XML необходим для использования в javascript datagird (JQGrid http://www.trirand.com/blog/)

Ответы [ 20 ]

2 голосов
/ 22 июля 2018

Хотя первоначальный вопрос возвращается некоторое время назад, количество ответов здесь (и мой собственный опыт) позволяют предположить, что это нетривиальный вопрос с множеством различных подходов различной сложности с различными компромиссами.

Вот почему я создал библиотеку SQLAthanor , которая расширяет декларативную ORM SQLAlchemy с настраиваемой поддержкой сериализации / десериализации, на которую вы можете взглянуть.

Библиотека поддерживает:

  • Python 2.7, 3.4, 3.5 и 3.6.
  • SQLAlchemy версии 0.9 и выше
  • сериализация / десериализация в / из JSON, CSV, YAML и Python dict
  • сериализация / десериализация столбцов / атрибутов, отношений, гибридных свойств и прокси-серверов ассоциаций
  • включение и отключение сериализации для определенных форматов и столбцов / отношений / атрибутов (например, вы хотитеподдерживать входящее password значение, но никогда не включать исходящий один)
  • обработка перед сериализацией и после десериализации (для проверки или приведения типов)
  • довольно простой синтаксис, который является и Pythonic и полностью совместим с собственным подходом SQLAlchemy

Вы можете проверить(я надеюсь!) подробные документы здесь: https://sqlathanor.readthedocs.io/en/latest

Надеюсь, это поможет!

2 голосов
/ 04 апреля 2016

Настраиваемая сериализация и десериализация.

"from_json" (метод класса) создает объект модели на основе данных json.

"десериализация" может вызываться только в экземпляре и объединять все данные из json в экземпляр модели.

"сериализация" - рекурсивная сериализация

__ write_only __ свойствонеобходим для определения свойств только для записи (например, "password_hash").

class Serializable(object):
    __exclude__ = ('id',)
    __include__ = ()
    __write_only__ = ()

    @classmethod
    def from_json(cls, json, selfObj=None):
        if selfObj is None:
            self = cls()
        else:
            self = selfObj
        exclude = (cls.__exclude__ or ()) + Serializable.__exclude__
        include = cls.__include__ or ()
        if json:
            for prop, value in json.iteritems():
                # ignore all non user data, e.g. only
                if (not (prop in exclude) | (prop in include)) and isinstance(
                        getattr(cls, prop, None), QueryableAttribute):
                    setattr(self, prop, value)
        return self

    def deserialize(self, json):
        if not json:
            return None
        return self.__class__.from_json(json, selfObj=self)

    @classmethod
    def serialize_list(cls, object_list=[]):
        output = []
        for li in object_list:
            if isinstance(li, Serializable):
                output.append(li.serialize())
            else:
                output.append(li)
        return output

    def serialize(self, **kwargs):

        # init write only props
        if len(getattr(self.__class__, '__write_only__', ())) == 0:
            self.__class__.__write_only__ = ()
        dictionary = {}
        expand = kwargs.get('expand', ()) or ()
        prop = 'props'
        if expand:
            # expand all the fields
            for key in expand:
                getattr(self, key)
        iterable = self.__dict__.items()
        is_custom_property_set = False
        # include only properties passed as parameter
        if (prop in kwargs) and (kwargs.get(prop, None) is not None):
            is_custom_property_set = True
            iterable = kwargs.get(prop, None)
        # loop trough all accessible properties
        for key in iterable:
            accessor = key
            if isinstance(key, tuple):
                accessor = key[0]
            if not (accessor in self.__class__.__write_only__) and not accessor.startswith('_'):
                # force select from db to be able get relationships
                if is_custom_property_set:
                    getattr(self, accessor, None)
                if isinstance(self.__dict__.get(accessor), list):
                    dictionary[accessor] = self.__class__.serialize_list(object_list=self.__dict__.get(accessor))
                # check if those properties are read only
                elif isinstance(self.__dict__.get(accessor), Serializable):
                    dictionary[accessor] = self.__dict__.get(accessor).serialize()
                else:
                    dictionary[accessor] = self.__dict__.get(accessor)
        return dictionary
1 голос
/ 05 мая 2016
def alc2json(row):
    return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])

Я думал, что поиграю в маленький гольф с этим.

К вашему сведению: я использую automap_base , поскольку у нас есть отдельно разработанная схема в соответствии с требованиями бизнеса.Я только что начал использовать SQLAlchemy сегодня, но в документации говорится, что automap_base является расширением декларативной_базы, которая представляется типичной парадигмой в ORM SQLAlchemy, поэтому я считаю, что это должно работать.per Решение Tjorriemorrie , но оно просто сопоставляет столбцы со значениями и обрабатывает типы Python с помощью str () - значений столбцов.Наши значения состоят из Python datetime.time и decimal.Decimal. Результаты типа класса, поэтому он выполняет свою работу.

Надеюсь, это поможет любому прохожему!

1 голос
/ 22 марта 2019

AlchemyEncoder замечательный, но иногда дает сбой с десятичными значениями.Вот улучшенный кодировщик, который решает десятичную проблему -

class AlchemyEncoder(json.JSONEncoder):
# To serialize SQLalchemy objects 
def default(self, obj):
    if isinstance(obj.__class__, DeclarativeMeta):
        model_fields = {}
        for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
            data = obj.__getattribute__(field)
            print data
            try:
                json.dumps(data)  # this will fail on non-encodable values, like other classes
                model_fields[field] = data
            except TypeError:
                model_fields[field] = None
        return model_fields
    if isinstance(obj, Decimal):
        return float(obj)
    return json.JSONEncoder.default(self, obj)
1 голос
/ 13 июня 2018

следующий код будет сериализовать результат sqlalchemy в json.

import json
from collections import OrderedDict


def asdict(self):
    result = OrderedDict()
    for key in self.__mapper__.c.keys():
        if getattr(self, key) is not None:
            result[key] = str(getattr(self, key))
        else:
            result[key] = getattr(self, key)
    return result


def to_array(all_vendors):
    v = [ ven.asdict() for ven in all_vendors ]
    return json.dumps(v) 

Призыв к веселью,

def all_products():
    all_products = Products.query.all()
    return to_array(all_products)
0 голосов
/ 08 июня 2016

Мой дубль, использующий (слишком много?) Словарей:

def serialize(_query):
    #d = dictionary written to per row
    #D = dictionary d is written to each time, then reset
    #Master = dictionary of dictionaries; the id Key (int, unique from database) 
    from D is used as the Key for the dictionary D entry in Master
    Master = {}
    D = {}
    x = 0
    for u in _query:
        d = u.__dict__
        D = {}
        for n in d.keys():
           if n != '_sa_instance_state':
                    D[n] = d[n]
        x = d['id']
        Master[x] = D
    return Master

Запуск с колбой (включая jsonify) и flask_sqlalchemy для печати выводов в формате JSON.

Вызов функции с помощью jsonify (serialize()).

Работает со всеми запросами SQLAlchemy, которые я пробовал до сих пор (работает SQLite3)

0 голосов
/ 05 февраля 2019

Встроенные дроссели сериализатора с utf-8 не могут декодировать недопустимый начальный байт для некоторых входов.Вместо этого я пошел с:

def row_to_dict(row):
    temp = row.__dict__
    temp.pop('_sa_instance_state', None)
    return temp


def rows_to_list(rows):
    ret_rows = []
    for row in rows:
        ret_rows.append(row_to_dict(row))
    return ret_rows


@website_blueprint.route('/api/v1/some/endpoint', methods=['GET'])
def some_api():
    '''
    /some_endpoint
    '''
    rows = rows_to_list(SomeModel.query.all())
    response = app.response_class(
        response=jsonplus.dumps(rows),
        status=200,
        mimetype='application/json'
    )
    return response
0 голосов
/ 13 мая 2016

Я знаю, что это довольно старый пост. Я принял решение, данное @SashaB, и изменил в соответствии с моими потребностями.

Я добавил следующие вещи к нему:

  1. Список игнорируемых полей: список полей, которые следует игнорировать при сериализации
  2. Список замены полей: словарь, содержащий имена полей, которые должны быть заменены значениями при сериализации.
  3. Удалены методы и сериализация BaseQuery

Мой код выглядит следующим образом:

def alchemy_json_encoder(revisit_self = False, fields_to_expand = [], fields_to_ignore = [], fields_to_replace = {}):
   """
   Serialize SQLAlchemy result into JSon
   :param revisit_self: True / False
   :param fields_to_expand: Fields which are to be expanded for including their children and all
   :param fields_to_ignore: Fields to be ignored while encoding
   :param fields_to_replace: Field keys to be replaced by values assigned in dictionary
   :return: Json serialized SQLAlchemy object
   """
   _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' and x not in fields_to_ignore]:
                val = obj.__getattribute__(field)
                # is this field method defination, or an SQLalchemy object
                if not hasattr(val, "__call__") and not isinstance(val, BaseQuery):
                    field_name = fields_to_replace[field] if field in fields_to_replace else 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_name] = None
                            continue

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

        return json.JSONEncoder.default(self, obj)
   return AlchemyEncoder

Надеюсь, это кому-нибудь поможет!

0 голосов
/ 13 декабря 2016

Используйте встроенный сериализатор в SQLAlchemy:

from sqlalchemy.ext.serializer import loads, dumps
obj = MyAlchemyObject()
# serialize object
serialized_obj = dumps(obj)

# deserialize object
obj = loads(serialized_obj)

Если вы переносите объект между сеансами, не забудьте отсоединить объект от текущего сеанса с помощью session.expunge(obj),Чтобы прикрепить его снова, просто сделайте session.add(obj).

0 голосов
/ 11 марта 2018

Под Flask это работает и обрабатывает поля времени данных, преобразуя поле типа
'time': datetime.datetime(2018, 3, 22, 15, 40) в
"time": "2018-03-22 15:40:00"

obj = {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}

# This to get the JSON body
return json.dumps(obj)

# Or this to get a response object
return jsonify(obj)
Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...