Jak serializować wynik SqlAlchemy do JSON?


192

Django ma dobrą automatyczną serializację modeli ORM zwróconych z DB do formatu JSON.

Jak serializować wynik zapytania SQLAlchemy do formatu JSON?

Próbowałem, jsonpickle.encodeale koduje sam obiekt zapytania. Próbowałem, json.dumps(items)ale powraca

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

Czy naprawdę tak trudno jest serializować obiekty ORM SQLAlchemy do JSON / XML? Czy nie ma dla niego domyślnego serializatora? Obecnie bardzo powszechne jest serializowanie wyników zapytań ORM.

Potrzebuję tylko zwrócić JSON lub XML reprezentację wyników zapytania SQLAlchemy.

Wynik zapytania o obiekty SQLAlchemy w formacie JSON / XML jest potrzebny do użycia w javascript datagird (JQGrid http://www.trirand.com/blog/ )


To obejście, które działa dla mnie. wprowadź opis linku tutaj
octaedro

Odpowiedzi:


129

Płaska realizacja

Możesz użyć czegoś takiego:

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)

a następnie przekonwertować na JSON, używając:

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

Zignoruje pola, które nie są kodowalne (ustaw je na „Brak”).

Nie rozszerza relacji automatycznie (ponieważ może to prowadzić do odwołań do siebie i zapętlać na zawsze).

Rekurencyjna, nieokrągła implementacja

Jeśli jednak wolisz zapętlać się w nieskończoność, możesz użyć:

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

A następnie koduj obiekty za pomocą:

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

Spowoduje to zakodowanie wszystkich dzieci i wszystkich ich dzieci oraz wszystkich ich dzieci ... Potencjalnie, w zasadzie, możesz zakodować całą bazę danych. Kiedy osiągnie coś, co wcześniej było zakodowane, zakoduje to jako „Brak”.

Rekurencyjna, być może cykliczna, selektywna implementacja

Inną alternatywą, prawdopodobnie lepszą, jest możliwość określenia pól, które chcesz rozwinąć:

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

Możesz teraz zadzwonić za pomocą:

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

Aby rozszerzyć tylko pola SQLAlchemy, na przykład „rodzice”.


że to świetna odpowiedź, jednak otrzymuję „może nie kodować«BaseQuery»ilekroć trafi relacji z metod innych niż płaskie, jakieś pomysły?
Ben Kilah

1
@SashaB Co powiesz na bardziej szczegółowe ukierunkowanie na przypadki, w których związek się powtarza? Na przykład, jeśli mam online_orderi addresszarówno związek z user, jak online_orderi związek z address. Gdybym chciał serializować to wszystko, musiałbym dołączyć addressdo fields_to_expand, ale nie chciałbym nadmiarowo serializować addressze względu na jego związek z jednym useri drugim online_order.
Chrispy

2
@BenKilah Niech zgadnę, używasz Flask-SqlAlchemy, a twoje modele dziedziczą po db.Model, a nie Base. W takim przypadku zmień go for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:tak, aby czytał for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata' and not x.startswith('query')]:. Należy pamiętać, że to rozwiązanie zapobiegnie posiadaniu własności / relacji z nazwą „zapytanie”
Pakman

tak samo jak ja, ale o wiele bardziej skomplikowane. stackoverflow.com/questions/7102754/...
tyan

2
Możesz użyć mojego rozwiązania github.com/n0nSmoker/SQLAlchemy-serializer
n0nSmoker

272

Możesz po prostu wyprowadzić obiekt jako słownik:

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

A potem używasz User.as_dict()do serializacji swojego obiektu.

Jak wyjaśniono w Konwertowanie obiektu wiersza sqlalchemy na Python Dict


2
@charlax, Jak mogłem naprawić DateTime? Korzystając z tego, otrzymuję „datetime.datetime (2013, 3, 22, 16, 50, 11) nie jest serializowalny w JSON”, gdy robię json.dumps
Asken

1
Jest to odpowiedzialność za JSONEncoderprzedmiot. Można go podklasować, aby zdefiniować własny koder dla niektórych obiektów, w tym datetime. Zwróć uwagę, że Flaskna przykład obsługuje kodowanie datetime w JSON po wyjęciu z pudełka (z najnowszą wersją).
charlax

3
Jeśli użyjesz metody „deklaratywnej” narzędzia sqlalchemy, możesz dodać coś takiego do niestandardowej klasy Base - jest to bardzo przydatne, ponieważ możesz wtedy wywołać my_orm_object.toDict () na dowolnym obiekcie ORM. Podobnie możesz zdefiniować metodę .toJSON (), która używa metody toDict i niestandardowego kodera do obsługi dat, obiektów blob itp.
FredL 15.04.13

7
w celu obsługi także datetime:return {c.name: unicode(getattr(self, c.name)) for c in self.__table__.columns}
Shoham

1
To nie działa, jeśli zmienne klas nie są takie same jak nazwy kolumn. Masz pomysł, jak zamiast tego uzyskać nazwy klas?
James Burke

55

Możesz przekonwertować RowProxy na taki słownik:

 d = dict(row.items())

Następnie serializuj to do JSON (będziesz musiał określić koder dla takich rzeczy jak datetimewartości). Nie jest to takie trudne, jeśli potrzebujesz tylko jednego rekordu (a nie pełnej hierarchii powiązanych rekordów).

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

1
Działa to dla mojego niestandardowego zapytania SQL z db.engine.connect () as con: rs = con.execute (sql)
JZ.

1
To jest o wiele prostsze i działa. Jaka jest różnica między tą odpowiedzią a odpowiedzią zaakceptowaną?
Niedziela

46

Polecam używanie pianki . Pozwala tworzyć serializatory reprezentujące instancje modelu z obsługą relacji i zagnieżdżonych obiektów.

Oto skrócony przykład z ich dokumentów. Weź model 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))

Schemat ptasie mleczko dla tej klasy jest zbudowany w następujący sposób:

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)

... i używane w ten sposób:

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

... wygeneruje taki wynik:

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

Zobacz ich pełny przykład Flask-SQLAlchemy .

Biblioteka o nazwie marshmallow-sqlalchemyspecjalnie integruje SQLAlchemy i piankę. W tej bibliotece schemat Authoropisanego powyżej modelu wygląda następująco:

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

Dzięki integracji typy pól można wywnioskować na podstawie Columntypów SQLAlchemy .

marshmallow-sqlalchemy tutaj.


12
Znalazłem też marshmallow-sqlalchemy.readthedocs.io/en/latest, który upraszcza generowanie schematu
Foo L

44

Python 3.7+ i Flask 1.1+ mogą korzystać z wbudowanego pakietu klas danych

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="user1@gmail.com"), User(email="user2@gmail.com")
  db.create_all()
  db.session.add_all(users)
  db.session.commit()
  app.run()

/users/Trasa będzie teraz powrócić do listy użytkowników.

[
  {"email": "user1@gmail.com", "id": 1},
  {"email": "user2@gmail.com", "id": 2}
]

Automatyczne serializowanie powiązanych modeli

@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

Odpowiedź jsonify(account)byłaby taka.

{  
   "id":1,
   "users":[  
      {  
         "email":"user1@gmail.com",
         "id":1
      },
      {  
         "email":"user2@gmail.com",
         "id":2
      }
   ]
}

Zastąp domyślny koder 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)
    elif type(o) == datetime.datetime:
      return o.isoformat()
    else:
      return super().default(o)

app.json_encoder = CustomJSONEncoder      

1
To wygląda na właściwy rodzaj prosty. Czy działa również w przypadku deserializacji?
Ender2050,

Możesz przekonwertować słownik parsowanego JSON na model, używając rozpakowywania argumentów słów kluczowych:data = request.json['user']; user = User(**data)
tom

3
Zauważ, że id: int = Columnzadziała, ale id = Columnnie, wydaje się, że MUSISZ zadeklarować wpisywanie statyczne, aby json zserializował pole, w przeciwnym razie otrzymasz pusty {}obiekt.
Ambroise Rabier,

1
To działało dla mnie, dlaczego nie jest to zaakceptowana odpowiedź? Godzinami bawię się w app_context, żeby to zadziałało z Flask-Marshmallow.
Nick Dat Le

1
Pracował również dla mnie. Należy pamiętać, że jeśli jesteś na Pythona 3.6, będziemy chcieli, aby po prostu zainstalować pakiet: pipenv install dataclasses. A potem wszystko będzie dobrze.
AleksandrH

14

Pakiet Flask-JsonTools ma implementację JsonSerializableBase klasy podstawowej dla twoich modeli.

Stosowanie:

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

Base = declarative_base(cls=(JsonSerializableBase,))

class User(Base):
    #...

Teraz Usermodel jest magicznie serializowalny.

Jeśli twój framework nie jest Flask, możesz po prostu pobrać kod


2
To rozwiązuje tylko połowę problemu, ponieważ serializuje tylko jeden wiersz. Jak serializować cały wynik zapytania?
Steve Bennett

@SteveBennett użyj jsonapi jsontools do zakodowania odpowiedzi. To automatycznie koduje zwracany obiekt
Tjorriemorrie

Mam bardzo prosty model sqlalchemy i otrzymuję: TypeError: <Obiekt ORM.State o wartości 0x03577A50> nie jest serializowany JSON
Matej

1
W końcu zadziałało, jawnie wywołując __json __ () na moim obiekcie modelu: return my_object .__ json __ ()
Matej

Biblioteka nie działa z Flask 1.0 i nowszymi, ponieważ import flask.ext.whatevernie jest już obsługiwana w Flask 1.0.
Adarsh ​​Madrecha

14

Ze względów bezpieczeństwa nigdy nie należy zwracać wszystkich pól modelu. Wolę je selektywnie wybierać.

Kodowanie json Flask obsługuje teraz UUID, datę queryi query_classgodzinę oraz relacje (i dodano oraz dla db.Modelklasy flask_sqlalchemy ). Zaktualizowałem koder w następujący sposób:

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

Dzięki temu mogę opcjonalnie dodać __json__właściwość, która zwraca listę pól, które chcę zakodować:

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

Dodaję @jsonapi do mojego widoku, zwracam listę wyników, a następnie moje wyniki są następujące:

[

{

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

]

Piękny! Jeszcze raz udowodnij, że czasami nie potrzebujesz grubego pakietu na każde głupie małe zadanie - że nauka DSL może być trudniejsza niż robienie tego „trudnym” sposobem. Zanim wylądowałem tutaj, przejrzałem wiele pakietów JSON i REST. To prawda, że ​​nadal wymaga pakietu flask_jsontools (aby dodać @jsonapido @app.routew views.py itp.), Ale uwielbiam jego prostotę. Myślę, że to tanie Data dodania kolby, ale nie data, więc dodałem ją sam do json_encoder.py : value=...^ if isinstance(value, date):^ data[field] = datetime.combine(value, time.min).isoformat()^ else:^try:...
juanitogan

10

Możesz użyć introspekcji SqlAlchemy w następujący sposób:

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

Zainspiruj się tutaj: Konwertuj obiekt wiersza sqlalchemy na Python Dict


5

Bardziej szczegółowe wyjaśnienie. W swoim modelu dodaj:

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

str()Jest pytona 3 tak, jeśli Pythonie 2 użycia unicode(). Powinno to pomóc w deserializacji dat. Możesz go usunąć, jeśli nie masz z nimi do czynienia.

Możesz teraz przeszukiwać bazę danych w ten sposób

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

First()jest potrzebny, aby uniknąć dziwnych błędów. as_dict()teraz deserializuje wynik. Po deserializacji jest gotowy do zmiany w Json

jsonify(some_result)

3

To nie jest takie proste. Napisałem kod, aby to zrobić. Nadal nad tym pracuję, i korzysta z frameworka MochiKit. Zasadniczo tłumaczy obiekty złożone między Pythonem i Javascript przy użyciu proxy i zarejestrowanych konwerterów JSON.

Stroną przeglądarki dla obiektów bazy danych jest db.js. Potrzebuje podstawowego źródła proxy Python w proxy.js .

Po stronie Pythona znajduje się podstawowy moduł proxy . Następnie koder obiektów SqlAlchemy w webserver.py . Zależy to również od ekstraktorów metadanych znajdujących się w pliku models.py .


Na pierwszy rzut oka dość skomplikowane ... Potrzebuję uzyskać wynik zapytania o obiekty SQLAlchemy w formacie JSON / XML, aby użyć go w javascript datagird (JQGrid trirand.com/blog )
Zelid

Czasami problemy są bardziej skomplikowane, niż się spodziewasz na pierwszy rzut oka ... Obsługuje obiekty zwrócone jako klucze obce i próbuje uniknąć nieskończonej rekurencji, która ma miejsce przy głęboko zagnieżdżonych relacjach. Jednak prawdopodobnie można napisać niestandardowe zapytania zwracające tylko typy podstawowe i serializować je bezpośrednio za pomocą simplejson.
Keith

1
Racja, może naprawdę pójdę z zapytaniem o dyktaty za pomocą SQLAlchemy i skorzystam z zalet ORM wykonujących tylko operacje zapisu / aktualizacji.
Zelid

3

Choć pierwotne pytanie się cofało, liczba odpowiedzi tutaj (i moich własnych doświadczeń) sugeruje, że jest to pytanie nietrywialne z wieloma różnymi podejściami o różnym stopniu złożoności z różnymi kompromisami.

Właśnie dlatego zbudowałem bibliotekę SQLAthanor, która rozszerza deklaratywną ORM SQLAlchemy o konfigurowalną obsługę serializacji / deserializacji , na którą warto spojrzeć.

Biblioteka obsługuje:

  • Python 2.7, 3.4, 3.5 i 3.6.
  • SQLAlchemy wersje 0.9 i wyższe
  • serializacja / de-serializacja do / z JSON, CSV, YAML i Python dict
  • serializacja / deserializacja kolumn / atrybutów, relacji, właściwości hybrydowych i serwerów proxy asocjacji
  • włączanie i wyłączanie serializacji dla określonych formatów i kolumn / relacji / atrybutów (np. chcesz obsługiwać wartość przychodzącą password , ale nigdy nie dołączaj wartości wychodzącej )
  • przetwarzanie wartości przed serializacją i po deserializacji (do walidacji lub przymusu typu)
  • dość prosta składnia, która jest zarówno Pythonic, jak i płynnie spójna z własnym podejściem SQLAlchemy

Możesz sprawdzić (mam nadzieję!) Obszerne dokumenty tutaj: https://sqlathanor.readthedocs.io/en/latest

Mam nadzieję że to pomoże!


2

Niestandardowa serializacja i deserializacja.

„from_json” (metoda klasy) buduje obiekt Model na podstawie danych json.

„deserializować” można wywołać tylko w instancji i scalić wszystkie dane z json w instancję Model.

„serialize” - serializacja rekurencyjna

Właściwość __write_only__ jest potrzebna do zdefiniowania właściwości tylko do zapisu (na przykład „ skrót_hasła ”).

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

2

Oto rozwiązanie, które pozwala wybrać relacje, które chcesz uwzględnić w swoich wynikach tak głęboko, jak chcesz. UWAGA: Jest to całkowite ponowne zapisanie, biorąc dict / str jako argument zamiast listy. naprawia niektóre rzeczy ..

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

więc na przykład używając osoby / rodziny / domu / pokoju ... zamieniając go w json, wszystko czego potrzebujesz to

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

To dobrze, myślę, że po prostu umieść w swojej klasie podstawowej, aby wszystkie obiekty miały to. Zostawię ci kodowanie Json ...
Tahoe

Zauważ, że ta wersja otrzyma wszystkie relacje list, więc bądź ostrożny, zapewniając stosunki z mnóstwem przedmiotów ...
tahoe

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

Myślałem, że zagram w ten golf z małym kodem.

FYI: Korzystam z automap_base, ponieważ mamy osobno zaprojektowany schemat zgodnie z wymaganiami biznesowymi. Właśnie zacząłem używać SQLAlchemy dzisiaj, ale dokumentacja stwierdza, że ​​automap_base jest rozszerzeniem do deklaratywnej_bazy, która wydaje się być typowym paradygmatem w SQLAlchemy ORM, więc uważam, że to powinno działać.

Nie nabiera ochoty śledzić kluczy obcych w rozwiązaniu Tjorriemorrie , ale po prostu dopasowuje kolumny do wartości i obsługuje typy Pythona poprzez str () - wartości kolumn. Nasze wartości składają się z wyników typu klasy Python datetime.time i decimal.Decimal, aby wykonać zadanie.

Mam nadzieję, że pomoże to przechodniom!


1

Wiem, że to dość starszy post. Wziąłem rozwiązanie podane przez @SashaB i zmodyfikowałem zgodnie z moimi potrzebami.

Dodałem do niego następujące rzeczy:

  1. Lista ignorowanych pól: lista pól, które należy zignorować podczas serializacji
  2. Lista zamiany pól: słownik zawierający nazwy pól, które mają być zastąpione wartościami podczas serializacji.
  3. Usunięte metody i BaseQuery są serializowane

Mój kod jest następujący:

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

Mam nadzieję, że to komuś pomaga!


1

Użyj wbudowanego serializatora w SQLAlchemy:

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

# deserialize object
obj = loads(serialized_obj)

Jeśli przenosisz obiekt między sesjami, pamiętaj, aby odłączyć obiekt od bieżącej sesji za pomocą session.expunge(obj). Aby dołączyć go ponownie, po prostu zrób session.add(obj).


Fajny, ale nie konwertuje na JSON.
blakev

2
Aby uzyskać „serializację” JSON, sprawdź marshmallow-sqlalchemy . Zdecydowanie najlepsze rozwiązanie, gdy eksponujesz obiekty klientom. marshmallow-sqlalchemy.readthedocs.io
chribsen

Moduł serializatora jest odpowiedni tylko dla struktur zapytań. Nie jest potrzebny do: instancji klas zdefiniowanych przez użytkownika. Nie zawierają one odniesień do mechanizmów, sesji ani konstrukcji wyrażeń w typowym przypadku i mogą być serializowane bezpośrednio.
thomasd

1

poniższy kod serializuje wynik sqlalchemy do 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) 

Wzywając zabawy

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

1

AlchemyEncoder jest wspaniały, ale czasami zawodzi przy wartościach dziesiętnych. Oto ulepszony koder, który rozwiązuje problem dziesiętny -

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

Gdy używasz sqlalchemy do łączenia się z db I, jest to proste rozwiązanie, które można w bardzo dużym stopniu konfigurować. Używaj pand.

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

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

3
Zrzuty kodu bez żadnego wyjaśnienia rzadko są pomocne. Przepełnienie stosu polega na uczeniu się, a nie udostępnianiu fragmentów do ślepego kopiowania i wklejania. Proszę edytować swoje pytanie i wyjaśnić, jak to działa lepiej niż to, co PO warunkiem.
Chris

0

Pod Flaskem działa i obsługuje pola czasu danych, przekształcając pole typu
'time': datetime.datetime(2018, 3, 22, 15, 40)w
"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)

0

Wbudowane dławiki serializatora z utf-8 nie mogą dekodować nieprawidłowego bajtu początkowego dla niektórych danych wejściowych. Zamiast tego poszedłem z:

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

Może możesz użyć takiej klasy

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!

Dzięki temu wszystkie obiekty mają to_dictmetodę


0

Podczas korzystania z niektórych surowych obiektów sql i niezdefiniowanych, cursor.descriptionpojawiało się używanie, aby uzyskać to, czego szukałem:

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)

-2

Moje podejście z wykorzystaniem (zbyt wielu?) Słowników:

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

Uruchamianie z flask (w tym jsonify) i flask_sqlalchemy, aby wydrukować wyniki jako JSON.

Wywołaj funkcję za pomocą jsonify (serialize ()).

Działa ze wszystkimi zapytaniami SQLAlchemy, które próbowałem do tej pory (działając SQLite3)

Korzystając z naszej strony potwierdzasz, że przeczytałeś(-aś) i rozumiesz nasze zasady używania plików cookie i zasady ochrony prywatności.
Licensed under cc by-sa 3.0 with attribution required.