Подтвердить что ты не робот

Как преобразовать результат SqlAlchemy в JSON?

Django имеет хорошую автоматическую сериализацию моделей ORM, возвращенных из DB в формат JSON.

Как сериализовать результат запроса SQLAlchemy в формате JSON?

Я пробовал jsonpickle.encode, но он сам кодирует объект запроса. Я попробовал json.dumps(items), но он возвращает

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

Неужели так сложно сериализовать объекты ORM SQLAlchemy в JSON/XML? Разве для этого нет сериализатора по умолчанию? Это очень распространенная задача для сериализации результатов запроса ORM в настоящее время.

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

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

4b9b3361

Ответ 1

Плоская реализация

Вы можете использовать что-то вроде этого:

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)

Он будет игнорировать поля, которые не кодируются (установите для них значение "Нет").

Он не расширяет отношения автоматически (поскольку это может привести к самоссылке и циклу навсегда).

Рекурсивная, некруглая реализация

Однако, если вы предпочитаете цикл навсегда, вы можете использовать:

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, называемые "родители".

Ответ 2

Вы можете просто вывести свой объект в виде словаря:

class User:
   def as_dict(self):
       return {c.name: getattr(self, c.name) for c in self.__table__.columns}

А затем вы используете User.as_dict() для сериализации вашего объекта.

Как объясняется в Преобразовать объект строки sqlalchemy в python dict

Ответ 3

Вы можете преобразовать RowProxy в dict так:

 d = dict(row.items())

Затем сериализуйте это для JSON (вам нужно будет указать кодировщик для таких вещей, как datetime) Это не так сложно, если вам просто нужна одна запись (а не полная иерархия связанных записей).

json.dumps([(dict(row.items())) for row in rs])

Ответ 4

Я рекомендую использовать зефир. Он позволяет создавать сериализаторы для представления экземпляров вашей модели с поддержкой отношений и вложенных объектов.

Вот усеченный пример из их документов. Возьмите модель ORM, Author:

class Author(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    first = db.Column(db.String(80))
    last = db.Column(db.String(80))

Схема зефира для этого класса строится так:

class AuthorSchema(Schema):
    id = fields.Int(dump_only=True)
    first = fields.Str()
    last = fields.Str()
    formatted_name = fields.Method("format_name", dump_only=True)

    def format_name(self, author):
        return "{}, {}".format(author.last, author.first)

... и используется так:

author_schema = AuthorSchema()
author_schema.dump(Author.query.first())

... выдаст такой результат:

{
        "first": "Tim",
        "formatted_name": "Peters, Tim",
        "id": 1,
        "last": "Peters"
}

Посмотрите их полный пример Flask-SQLAlchemy.

Библиотека под названием marshmallow-sqlalchemy специально объединяет SQLAlchemy и зефир. В этой библиотеке схема для модели Author, описанная выше, выглядит следующим образом:

class AuthorSchema(ModelSchema):
    class Meta:
        model = Author

Интеграция позволяет выводить типы полей из типов SQLAlchemy Column.

Зефир-sqlalchemy здесь.

Ответ 5

Пакет Flask-JsonTools имеет реализацию JsonSerializableBase Базовый класс для ваших моделей.

Использование:

from sqlalchemy.ext.declarative import declarative_base
from flask.ext.jsontools import JsonSerializableBase

Base = declarative_base(cls=(JsonSerializableBase,))

class User(Base):
    #...

Теперь модель User магически сериализуема.

Если ваша фреймворк не Flask, вы можете просто захватить код

Ответ 6

По соображениям безопасности вы никогда не должны возвращать все поля модели. Я предпочитаю выборочно выбирать их.

Кодирование flask json теперь поддерживает UUID, datetime и отношения (и добавлено query и query_class для класса flask_sqlalchemy db.Model). Я обновил кодировщик следующим образом:

приложение/json_encoder.py

    from sqlalchemy.ext.declarative import DeclarativeMeta
    from flask import json


    class AlchemyEncoder(json.JSONEncoder):
        def default(self, o):
            if isinstance(o.__class__, DeclarativeMeta):
                data = {}
                fields = o.__json__() if hasattr(o, '__json__') else dir(o)
                for field in [f for f in fields if not f.startswith('_') and f not in ['metadata', 'query', 'query_class']]:
                    value = o.__getattribute__(field)
                    try:
                        json.dumps(value)
                        data[field] = value
                    except TypeError:
                        data[field] = None
                return data
            return json.JSONEncoder.default(self, o)

app/__init__.py

# json encoding
from app.json_encoder import AlchemyEncoder
app.json_encoder = AlchemyEncoder

С этим я могу дополнительно добавить свойство __json__, которое возвращает список полей, которые я хочу кодировать:

app/models.py

class Queue(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    song_id = db.Column(db.Integer, db.ForeignKey('song.id'), unique=True, nullable=False)
    song = db.relationship('Song', lazy='joined')
    type = db.Column(db.String(20), server_default=u'audio/mpeg')
    src = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime, server_default=db.func.now())
    updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now())

    def __init__(self, song):
        self.song = song
        self.src = song.full_path

    def __json__(self):
        return ['song', 'src', 'type', 'created_at']

Я добавляю @jsonapi к моему представлению, возвращаю результирующий список, а затем мой вывод выглядит следующим образом:

[

{

    "created_at": "Thu, 23 Jul 2015 11:36:53 GMT",
    "song": 

        {
            "full_path": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3",
            "id": 2,
            "path_name": "Audioslave/Audioslave [2002]/1 Cochise.mp3"
        },
    "src": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3",
    "type": "audio/mpeg"
}

]

Ответ 7

Вы можете использовать интроспекцию SqlAlchemy следующим образом:

mysql = SQLAlchemy()
from sqlalchemy import inspect

class Contacts(mysql.Model):  
    __tablename__ = 'CONTACTS'
    id = mysql.Column(mysql.Integer, primary_key=True)
    first_name = mysql.Column(mysql.String(128), nullable=False)
    last_name = mysql.Column(mysql.String(128), nullable=False)
    phone = mysql.Column(mysql.String(128), nullable=False)
    email = mysql.Column(mysql.String(128), nullable=False)
    street = mysql.Column(mysql.String(128), nullable=False)
    zip_code = mysql.Column(mysql.String(128), nullable=False)
    city = mysql.Column(mysql.String(128), nullable=False)
    def toDict(self):
        return { c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs }

@app.route('/contacts',methods=['GET'])
def getContacts():
    contacts = Contacts.query.all()
    contactsArr = []
    for contact in contacts:
        contactsArr.append(contact.toDict()) 
    return jsonify(contactsArr)

@app.route('/contacts/<int:id>',methods=['GET'])
def getContact(id):
    contact = Contacts.query.get(id)
    return jsonify(contact.toDict())

Получите вдохновение от ответа здесь:   Преобразование объекта строки sqlalchemy в python dict

Ответ 8

Более подробное объяснение. В вашей модели добавьте:

def as_dict(self):
       return {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}

str() для python 3, поэтому, если использовать python 2, используйте unicode(). Это должно помочь десериализовать даты. Вы можете удалить его, если не имеете дело с ними.

Теперь вы можете запросить базу данных следующим образом:

some_result = User.query.filter_by(id=current_user.id).first().as_dict()

First() необходим, чтобы избежать странных ошибок. as_dict() теперь десериализует результат. После десериализации он готов к обращению к json

jsonify(some_result)

Ответ 9

Python 3. 7+ и Flask 1. 1+ могут использовать встроенный пакет данных package

from dataclasses import dataclass
from datetime import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy


app = Flask(__name__)
db = SQLAlchemy(app)


@dataclass
class User(db.Model):
  id: int
  email: str

  id = db.Column(db.Integer, primary_key=True, auto_increment=True)
  email = db.Column(db.String(200), unique=True)


@app.route('/users/')
def users():
  users = User.query.all()
  return jsonify(users)  


if __name__ == "__main__":
  users = User(email="[email protected]"), User(email="[email protected]")
  db.create_all()
  db.session.add_all(users)
  db.session.commit()
  app.run()

Маршрут /users/ теперь возвращает список пользователей.

[
  {"email": "[email protected]", "id": 1},
  {"email": "[email protected]", "id": 2}
]

Автосериализация связанных моделей

@dataclass
class Account(db.Model):
  id: int
  users: User

  id = db.Column(db.Integer)
  users = db.relationship(User)  # User model would need a db.ForeignKey field

Ответ от jsonify(account) будет таким.

{  
   "id":1,
   "users":[  
      {  
         "email":"[email protected]",
         "id":1
      },
      {  
         "email":"[email protected]",
         "id":2
      }
   ]
}

Перезаписать кодер JSON по умолчанию

from flask.json import JSONEncoder


class CustomJSONEncoder(JSONEncoder):
  "Add support for serializing timedeltas"

  def default(o):
    if type(o) == datetime.timedelta:
      return str(o)
    else:
      return super().default(o)

app.json_encoder = CustomJSONEncoder      

Ответ 10

Это не так страшно. Я написал код для этого. Я все еще работаю над этим, и он использует структуру MochiKit. Он в основном переводит сложные объекты между Python и Javascript с использованием прокси и зарегистрированных конвертеров JSON.

Сторона браузера для объектов базы данных db.js Для этого нужен базовый источник прокси-сервера Python в proxy.js.

На стороне Python есть базовый прокси-модуль. Затем, наконец, кодер объекта SqlAlchemy в webserver.py. Это также зависит от экстрактов метаданных, найденных в файле models.py.

Ответ 11

Пользовательская сериализация и десериализация.

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

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

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

__ 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

Ответ 12

Вот решение, которое позволяет вам выбрать отношения, которые вы хотите включить в свой вывод, настолько глубоко, насколько хотите. ПРИМЕЧАНИЕ. Это полная переписывание с использованием dict/str в качестве аргумента, а не списка. исправляет некоторые вещи..

def deep_dict(self, relations={}):
    """Output a dict of an SA object recursing as deep as you want.

    Takes one argument, relations which is a dictionary of relations we'd
    like to pull out. The relations dict items can be a single relation
    name or deeper relation names connected by sub dicts

    Example:
        Say we have a Person object with a family relationship
            person.deep_dict(relations={'family':None})
        Say the family object has homes as a relation then we can do
            person.deep_dict(relations={'family':{'homes':None}})
            OR
            person.deep_dict(relations={'family':'homes'})
        Say homes has a relation like rooms you can do
            person.deep_dict(relations={'family':{'homes':'rooms'}})
            and so on...
    """
    mydict =  dict((c, str(a)) for c, a in
                    self.__dict__.items() if c != '_sa_instance_state')
    if not relations:
        # just return ourselves
        return mydict

    # otherwise we need to go deeper
    if not isinstance(relations, dict) and not isinstance(relations, str):
        raise Exception("relations should be a dict, it is of type {}".format(type(relations)))

    # got here so check and handle if we were passed a dict
    if isinstance(relations, dict):
        # we were passed deeper info
        for left, right in relations.items():
            myrel = getattr(self, left)
            if isinstance(myrel, list):
                mydict[left] = [rel.deep_dict(relations=right) for rel in myrel]
            else:
                mydict[left] = myrel.deep_dict(relations=right)
    # if we get here check and handle if we were passed a string
    elif isinstance(relations, str):
        # passed a single item
        myrel = getattr(self, relations)
        left = relations
        if isinstance(myrel, list):
            mydict[left] = [rel.deep_dict(relations=None)
                                 for rel in myrel]
        else:
            mydict[left] = myrel.deep_dict(relations=None)

    return mydict

поэтому для примера, используя человека/семью/дома/комнаты... превращая его в json, все, что вам нужно,

json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))

Ответ 13

В то время как исходный вопрос повторяется некоторое время, количество ответов здесь (и моих собственных опытов) предлагает ему нетривиальный вопрос с множеством различных подходов различной сложности с различными компромиссами.

Вот почему я построил библиотеку 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

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

Ответ 14

def alc2json(row):
    return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])

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

FYI: Я использую automap_base, так как у нас есть отдельно разработанная схема в соответствии с бизнес-требованиями. Я только начал использовать SQLAlchemy сегодня, но в документации указано, что automap_base является расширением для declarative_base, который, кажется, является типичной парадигмой ORM SQLAlchemy, поэтому я считаю, что это должно работать.

Это не похоже на наличие внешних ключей в Tjorriemorrie, но оно просто сопоставляет столбцы со значениями и обрабатывает типы Python с помощью str() - значения столбца. Наши значения состоят из Python datetime.time и decimal. Результат типа класса, поэтому он выполняет задание.

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

Ответ 15

Используйте встроенный сериализатор в 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).

Ответ 16

следующий код будет сериализовать результат 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)

Ответ 17

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)

Ответ 18

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

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

  • Список игнорирования полей: список полей, которые следует игнорировать при сериализации
  • Список замены поля: словарь, содержащий имена полей, которые будут заменены значениями при сериализации.
  • Удаленные методы и получение 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

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

Ответ 19

Под флягой это работает и обрабатывает поля данных, преобразуя поле типа
'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)

Ответ 20

Встроенные дроссели сериализатора с 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

Ответ 21

Мой прием использует (слишком много?) словарей:

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)