Django a une bonne sérialisation automatique des modèles ORM retournés du format DB au format JSON.

Comment sérialiser le résultat de la requête SQLAlchemy au format JSON?

J'ai essayé jsonpickle.encodemais il encode l'objet de requête lui-même. J'ai essayé json.dumps(items)mais ça revient

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

Est-il vraiment si difficile de sérialiser des objets SQLAlchemy ORM en JSON / XML? N'y a-t-il pas de sérialiseur par défaut pour cela? De nos jours, il est très courant de sérialiser les résultats des requêtes ORM.

Ce dont j'ai besoin est simplement de renvoyer une représentation de données JSON ou XML du résultat de la requête SQLAlchemy.

Le résultat de la requête des objets SQLAlchemy au format JSON / XML est nécessaire pour être utilisé dans javascript datagird (JQGrid http://www.trirand.com/blog/ )

Une mise en œuvre plate

Vous pouvez utiliser quelque chose comme ceci:

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)

puis convertissez en JSON en utilisant:

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

Il ignorera les champs qui ne sont pas encodables (définissez-les sur «Aucun»).

Il ne développe pas automatiquement les relations (car cela pourrait conduire à des auto-références et boucler pour toujours).

Une implémentation récursive et non circulaire

Si, cependant, vous préférez boucler indéfiniment, vous pouvez utiliser:

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

Et puis encodez les objets en utilisant:

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

Cela encoderait tous les enfants, et tous leurs enfants, et tous leurs enfants ... Potentiellement, encoder toute votre base de données, en gros. Quand il atteint quelque chose de son encodé avant, il l'encodera comme «Aucun».

Une implémentation sélective récursive, éventuellement circulaire

Une autre alternative, probablement meilleure, est de pouvoir spécifier les champs que vous souhaitez développer:

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

Vous pouvez maintenant l'appeler avec:

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

Pour développer uniquement les champs SQLAlchemy appelés «parents», par exemple.

Vous pouvez simplement afficher votre objet sous forme de dictionnaire:

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

Et puis vous utilisez User.as_dict()pour sérialiser votre objet.

Comme expliqué dans Convertir l'objet de ligne sqlalchemy en dict python

Python 3.7+ et Flask 1.1+ peuvent utiliser le package de classes de données intégré

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()

L' /users/itinéraire renverra maintenant une liste d'utilisateurs.

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

Sérialisation automatique des modèles associés

@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

La réponse jsonify(account)serait la suivante.

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

Remplacer l'encodeur JSON par défaut

from flask.json import JSONEncoder


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

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

app.json_encoder = CustomJSONEncoder      

Vous pouvez convertir un RowProxy en un dict comme ceci:

 d = dict(row.items())

Ensuite, sérialisez cela en JSON (vous devrez spécifier un encodeur pour des choses comme des datetimevaleurs) Ce n'est pas si difficile si vous ne voulez qu'un seul enregistrement (et non une hiérarchie complète d'enregistrements associés).

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

Je recommande d'utiliser la guimauve . Il vous permet de créer des sérialiseurs pour représenter vos instances de modèle avec la prise en charge des relations et des objets imbriqués.

Voici un exemple tronqué de leurs documents. Prenez le modèle 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))

Un schéma de guimauve pour cette classe est construit comme ceci:

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)

... et utilisé comme ceci:

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

... produirait une sortie comme celle-ci:

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

Jetez un œil à leur exemple complet Flask-SQLAlchemy .

Une bibliothèque appelée marshmallow-sqlalchemyspécifiquement intègre SQLAlchemy et marshmallow. Dans cette bibliothèque, le schéma du Authormodèle décrit ci-dessus ressemble à ceci:

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

L'intégration permet aux types de champ d'être déduits des Columntypes SQLAlchemy .

guimauve-sqlalchemy ici.

Le package Flask-JsonTools a une implémentation de la classe JsonSerializableBase Base pour vos modèles.

Usage:

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

Base = declarative_base(cls=(JsonSerializableBase,))

class User(Base):
    #...

Maintenant, le Usermodèle est sérialisable par magie.

Si votre framework n'est pas Flask, vous pouvez simplement récupérer le code

Pour des raisons de sécurité, vous ne devez jamais renvoyer tous les champs du modèle. Je préfère les choisir de manière sélective.

Le codage json de Flask prend désormais en charge l'UUID, la date / heure et les relations (et ajouté queryet query_classpour la db.Modelclasse flask_sqlalchemy ). J'ai mis à jour l'encodeur comme suit:

app / 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

Avec cela, je peux éventuellement ajouter une __json__propriété qui renvoie la liste des champs que je souhaite encoder:

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']

J'ajoute @jsonapi à ma vue, je renvoie la liste de résultats et ma sortie est la suivante:

[

{

    "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"
}

]

Vous pouvez utiliser l'introspection de SqlAlchemy comme ceci:

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())

Inspirez- vous d'une réponse ici: Convertissez l'objet de ligne sqlalchemy en dict python

Une explication plus détaillée. Dans votre modèle, ajoutez:

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

Le str()est pour python 3 donc si vous utilisez python 2, utilisez unicode(). Cela devrait aider à désérialiser les dates. Vous pouvez le supprimer si vous ne vous en occupez pas.

Vous pouvez maintenant interroger la base de données comme ceci

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

First()est nécessaire pour éviter des erreurs étranges. as_dict()va maintenant désérialiser le résultat. Après la désérialisation, il est prêt à être transformé en json

jsonify(some_result)

Ce n'est pas si simple. J'ai écrit du code pour cela. J'y travaille toujours, et il utilise le framework MochiKit. Il traduit essentiellement des objets composés entre Python et Javascript à l'aide d'un proxy et de convertisseurs JSON enregistrés.

Le côté navigateur pour les objets de base de données est db.js. Il a besoin de la source proxy Python de base dans proxy.js .

Du côté Python, il y a le module proxy de base . Puis enfin l'encodeur d'objets SqlAlchemy dans webserver.py . Cela dépend également des extracteurs de métadonnées trouvés dans le fichier models.py .

Bien que la question originale remonte à un certain temps, le nombre de réponses ici (et mes propres expériences) suggèrent que c'est une question non triviale avec beaucoup d'approches différentes de complexité variable avec différents compromis.

C'est pourquoi j'ai créé la bibliothèque SQLAthanor qui étend l'ORM déclaratif de SQLAlchemy avec une prise en charge configurable de la sérialisation / désérialisation que vous voudrez peut-être examiner.

La bibliothèque prend en charge:

• Python 2.7, 3.4, 3.5 et 3.6.
• SQLAlchemy versions 0.9 et supérieures
• sérialisation / désérialisation vers / depuis JSON, CSV, YAML et Python dict
• sérialisation / désérialisation des colonnes / attributs, relations, propriétés hybrides et proxys d'association
• activation et désactivation de la sérialisation pour des formats et des colonnes / relations / attributs particuliers (par exemple, vous voulez prendre en charge un password valeur entrante , mais ne jamais inclure une valeur sortante )
• traitement de la valeur de pré-sérialisation et de post-désérialisation (pour validation ou coercition de type)
• une syntaxe assez simple qui est à la fois pythonique et parfaitement cohérente avec la propre approche de SQLAlchemy

Vous pouvez consulter la documentation complète (j'espère!) Ici: https://sqlathanor.readthedocs.io/en/latest

J'espère que cela t'aides!

Sérialisation et désérialisation personnalisées.

"from_json" (méthode de classe) construit un objet Model basé sur des données json.

"désérialiser" ne pouvait être appelé que sur l'instance et fusionner toutes les données de json dans l'instance de modèle.

"sérialiser" - sérialisation récursive

La propriété __write_only__ est nécessaire pour définir les propriétés en écriture seule ("password_hash" par exemple).

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

Utilisez le sérialiseur intégré dans SQLAlchemy:

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

# deserialize object
obj = loads(serialized_obj)

Si vous transférez l'objet entre les sessions, n'oubliez pas de détacher l'objet de la session en cours à l'aide de session.expunge(obj). Pour le rattacher, faites-le session.add(obj).

Voici une solution qui vous permet de sélectionner les relations que vous souhaitez inclure dans votre sortie aussi profondément que vous le souhaitez. REMARQUE: Il s'agit d'une réécriture complète prenant un dict / str comme argument plutôt que comme une liste. corrige quelques trucs ..

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

Donc, pour un exemple utilisant personne / famille / maisons / pièces ... le transformer en json, tout ce dont vous avez besoin est

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

step1:
class CNAME:
   ...
   def as_dict(self):
       return {item.name: getattr(self, item.name) for item in self.__table__.columns}

step2:
list = []
for data in session.query(CNAME).all():
    list.append(data.as_dict())

step3:
return jsonify(list)

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

Je pensais jouer un petit code de golf avec celui-ci.

FYI: J'utilise automap_base car nous avons un schéma conçu séparément en fonction des besoins de l'entreprise. Je viens de commencer à utiliser SQLAlchemy aujourd'hui, mais la documentation indique que automap_base est une extension de declarative_base qui semble être le paradigme typique de SQLAlchemy ORM, donc je pense que cela devrait fonctionner.

Il n'est pas compliqué de suivre les clés étrangères par Tjorriemorrie la solution de , mais il fait simplement correspondre les colonnes aux valeurs et gère les types Python en str () --ing les valeurs de colonne. Nos valeurs sont composées de résultats de type Python datetime.time et decimal.Decimal, ce qui permet de faire le travail.

J'espère que cela aidera tous les passants!

Je sais que c'est un article assez ancien. J'ai pris la solution donnée par @SashaB et modifiée selon mon besoin.

J'ai ajouté les éléments suivants:

1. Liste des champs ignorés: une liste de champs à ignorer lors de la sérialisation
2. Liste de remplacement de champ: un dictionnaire contenant les noms de champ à remplacer par des valeurs lors de la sérialisation.
3. Méthodes supprimées et BaseQuery en cours de sérialisation

Mon code est le suivant:

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

J'espère que cela aide quelqu'un!

le code suivant sérialisera le résultat de sqlalchemy en 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) 

Appel amusant,

def all_products():
    all_products = Products.query.all()
    return to_array(all_products)

L'AlchemyEncoder est merveilleux mais échoue parfois avec des valeurs décimales. Voici un encodeur amélioré qui résout le problème décimal -

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)

Lorsque vous utilisez sqlalchemy pour vous connecter à une base de données I, il s'agit d'une solution simple hautement configurable. Utilisez des pandas.

import pandas as pd
import sqlalchemy

#sqlalchemy engine configuration
engine = sqlalchemy.create_engine....

def my_function():
  #read in from sql directly into a pandas dataframe
  #check the pandas documentation for additional config options
  sql_DF = pd.read_sql_table("table_name", con=engine)

  # "orient" is optional here but allows you to specify the json formatting you require
  sql_json = sql_DF.to_json(orient="index")

  return sql_json

installer simplejson par pip install simplejsonet créer une classe

class Serialise(object):

    def _asdict(self):
        """
        Serialization logic for converting entities using flask's jsonify

        :return: An ordered dictionary
        :rtype: :class:`collections.OrderedDict`
        """

        result = OrderedDict()
        # Get the columns
        for key in self.__mapper__.c.keys():
            if isinstance(getattr(self, key), datetime):
                result["x"] = getattr(self, key).timestamp() * 1000
                result["timestamp"] = result["x"]
            else:
                result[key] = getattr(self, key)

        return result

et héritez de cette classe pour toutes les classes orm afin que cette _asdictfonction soit enregistrée dans chaque classe ORM et boom. Et utilisez jsonify n'importe où

Sous Flask, cela fonctionne et gère les champs datatime, transformant un champ de type
'time': datetime.datetime(2018, 3, 22, 15, 40)en
"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)

Les selfs de sérialiseur intégrés avec utf-8 ne peuvent pas décoder l'octet de démarrage invalide pour certaines entrées. Au lieu de cela, je suis allé avec:

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

Peut-être que vous pouvez utiliser une classe comme celle-ci

from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy import Table


class Custom:
    """Some custom logic here!"""

    __table__: Table  # def for mypy

    @declared_attr
    def __tablename__(cls):  # pylint: disable=no-self-argument
        return cls.__name__  # pylint: disable= no-member

    def to_dict(self) -> Dict[str, Any]:
        """Serializes only column data."""
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}

Base = declarative_base(cls=Custom)

class MyOwnTable(Base):
    #COLUMNS!

Avec cela tous les objets ont la to_dictméthode

Lors de l'utilisation de certains SQL bruts et d'objets non définis, l'utilisation cursor.descriptionsemblait obtenir ce que je cherchais:

with connection.cursor() as cur:
    print(query)
    cur.execute(query)
    for item in cur.fetchall():
        row = {column.name: item[i] for i, column in enumerate(cur.description)}
        print(row)

Même si c'est un ancien post, peut-être que je n'ai pas répondu à la question ci-dessus, mais je veux parler de ma sérialisation, au moins cela fonctionne pour moi.

J'utilise FastAPI, SqlAlchemy et MySQL, mais je n'utilise pas le modèle orm;

# from sqlalchemy import create_engine
# from sqlalchemy.orm import sessionmaker
# engine = create_engine(config.SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
# SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Code de sérialisation

import decimal
import datetime


def alchemy_encoder(obj):
    """JSON encoder function for SQLAlchemy special classes."""
    if isinstance(obj, datetime.date):
        return obj.strftime("%Y-%m-%d %H:%M:%S")
    elif isinstance(obj, decimal.Decimal):
        return float(obj)

import json
from sqlalchemy import text

# db is SessionLocal() object 

app_sql = 'SELECT * FROM app_info ORDER BY app_id LIMIT :page,:page_size'

# The next two are the parameters passed in
page = 1
page_size = 10

# execute sql and return a <class 'sqlalchemy.engine.result.ResultProxy'> object
app_list = db.execute(text(app_sql), {'page': page, 'page_size': page_size})

# serialize
res = json.loads(json.dumps([dict(r) for r in app_list], default=alchemy_encoder))

Si cela ne fonctionne pas, veuillez ignorer ma réponse. Je me réfère ici

https://codeandlife.com/2014/12/07/sqlalchemy-results-to-json-the-easy-way/

Ma prise en utilisant (trop?) De dictionnaires:

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

Exécution avec flask (y compris jsonify) et flask_sqlalchemy pour imprimer les sorties au format JSON.

Appelez la fonction avec jsonify (serialize ()).

Fonctionne avec toutes les requêtes SQLAlchemy que j'ai essayées jusqu'à présent (exécutant SQLite3)