SqlAlchemy 결과를 JSON으로 시리얼화하는 방법

SqlAlchemy 결과를 JSON으로 시리얼화하는 방법

Django는 DB 형식에서 JSON 형식으로 반환되는 ORM 모델의 자동 시리얼라이즈를 갖추고 있습니다.

SQL Chemy 쿼리 결과를 JSON 형식으로 어떻게 시리얼화합니까?

는 는 i i는노노 i i i i i.jsonpickle.encode쿼리 객체 자체를 인코딩합니다.는 는 i i는노노 i i i i i.json.dumps(items) 돌아오다

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

SQL Chemy ORM 개체를 JSON/XML로 직렬화하는 것이 그렇게 어렵습니까? 기본 직렬화기는 없습니까?오늘날 ORM 쿼리 결과를 직렬화하는 작업은 매우 일반적입니다.

SQL Chemy 쿼리 결과의 JSON 또는 XML 데이터 표현만 반환하면 됩니다.

javascript datagird(JQGrid http://www.trirand.com/blog/))에서 사용하려면 JSON/XML 형식의 SQL Chemy 객체 쿼리 결과가 필요합니다.

개체를 사전으로 출력할 수 있습니다.

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

에 리고 and and and and를 사용합니다.User.as_dict()를 클릭하여 개체를 직렬화합니다.

설명에 따라 sqlalchemy 행 개체를 python dict로 변환

플랫 실장

다음과 같은 것을 사용할 수 있습니다.

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)

예를 들어 'parents'라고 하는 SQL Chemy 필드만 확장합니다.

Python 3.7+와 Flask 1.1+는 내장 데이터 클래스 패키지를 사용할 수 있습니다.

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/【루트】

[
  {"email": "user1@gmail.com", "id": 1},
  {"email": "user2@gmail.com", "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":"user1@gmail.com",
         "id":1
      },
      {  
         "email":"user2@gmail.com",
         "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)
    elif type(o) == datetime.datetime:
      return o.isoformat()
    else:
      return super().default(o)

app.json_encoder = CustomJSONEncoder      

RowProxy를 다음과 같은 딕트로 변환할 수 있습니다.

 d = dict(row.items())

으로 JSON에(JSON의 인코더를 ).datetimevalues 는 1개의 레코드의 로 하는 어렵지 않습니다.values ) 1개의 레코드(values)를 사용합니다.

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

마시멜로 사용을 추천합니다.관계 및 중첩된 객체를 지원하는 모델 인스턴스(instance)를 나타내는 시리얼라이저를 작성할 수 있습니다.

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

풀 플라스크-SQ를 보세요.Lalchemy의 예.

고는 called called called 라는 marshmallow-sqlalchemySQL Chemy 와my che 。 라이브러리에서 " " "의 "Author위의 모델은 다음과 같습니다.

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

통합을 통해 SQL 할 수 .Columntypes.timeout.

마시멜로-스칼케미.

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로 변환합니다.

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가 아니라면, 당신은 코드를 잡을 수 있습니다.

보안상의 이유로 모델의 모든 필드를 반환해서는 안 됩니다.나는 그것들을 선별적으로 고르는 것을 선호한다.

UUID, 및 Flask' json UUID, 짜및및및및및및 ( 및및및및및및및) 。query ★★★★★★★★★★★★★★★★★」query_classdb.Model같이 인코더를 다음과 같이 업데이트했습니다.

app/json_displays.화이

    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__"CHANGE: "CHANGE: "CHANGE: " 。

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

]

좀 더 자세히 설명해 주세요.모델에서 다음을 추가합니다.

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

str()32 를 는 python 3 을 사용합니다.unicode() 문제에할 수 .그것들을 처리하지 않으면 제거할 수 있습니다.

이렇게 데이터베이스를 쿼리할 수 있습니다.

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

First()이치노 as_dict()이치노할 수 .

jsonify(some_result)

원래의 질문은 꽤 오래되었지만, 여기(그리고 나의 경험도)의 회답의 수에 의하면, 이 질문은, 여러가지 복잡한 접근방식과 다양한 트레이드 오프를 수반하는, 지극히 간단한 질문입니다.

그래서 SQL Chemy의 선언적 ORM을 확장한 SQL Thanor 라이브러리를 구축했습니다.이것에 의해, 시리얼화/디시리얼라이제이션의 설정 가능한 서포트를 참조할 수 있습니다.

라이브러리는 다음을 지원합니다.

• Python 2.7, 3.4, 3.5 및 3.6.
• SQL Chemy 버전 0.9 이후
• Python JSON, CSV, YAML "Python"의 dict
• 열/속성의 직렬화/비활성화, 관계, 하이브리드 속성 및 연결 프록시
• 특정 형식 및 열/관계/속성에 대한 직렬화 활성화 및 비활성화(예: 인바운드 지원) password발신값은 포함하지 않습니다.)
• (검증 또는 유형 강제의 경우) 사전 평가 및 사후 평가 값 처리
• 피토닉어이며 SQL Chemy 자신의 접근법과 심리스하게 일치하는 매우 간단한 구문

(기대되는!) 포괄적인 문서는 다음 URL에서 확인하실 수 있습니다.https://sqlathanor.readthedocs.io/en/latest

이게 도움이 됐으면 좋겠네요!

커스텀 시리얼화 및 디시리얼화

"from_json"(클래스 메서드)은 json 데이터를 기반으로 Model 개체를 작성합니다.

"deserialize"는 인스턴스에서만 호출할 수 있으며 json의 모든 데이터를 모델 인스턴스로 병합합니다.

"재귀화" - 재귀적 시리얼화

__write_only__ 속성은 쓰기 전용 속성("password_only")을 정의하기 위해 필요합니다.

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

SQL Chemy에 내장된 시리얼라이저를 사용합니다.

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

다음은 출력에 포함할 관계를 원하는 만큼 선택할 수 있는 솔루션입니다.메모: 이것은 목록이 아닌 arg로 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'}}))

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)

오래된 투고이지만, 위의 질문에 대답하지 않은 것 같습니다만, 제 연재물에 대해 이야기하고 싶습니다만, 적어도 저는 효과가 있습니다.

FastAPI, SqlAlchemy 및 MySQL을 사용하지만 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)

시리얼화 코드

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

안 되면 제 답변을 무시해 주세요.여기서 언급하고 있습니다.

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

을 simplejson으로 pip install simplejson.

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

하고 이 클래스는 이 클래스가 하겠습니다._asdict함수가 모든 ORM 클래스에 등록되고 붐이 발생합니다. 임의의 에서 jsonify를 합니다.

그것은 그렇게 간단하지 않다.이걸 하기 위해 코드를 몇 개 썼어요.아직 작업 중이고, MochiKit 프레임워크를 사용하고 있습니다.기본적으로 프록시와 등록된 JSON 컨버터를 사용하여 Python과 Javascript 간의 복합 객체를 번역합니다.

데이터베이스 오브젝트의 브라우저측은 db.js 입니다.프록시.js에 기본 Python 프록시 소스가 필요합니다.

Python측에는 베이스 프록시 모듈이 있습니다.마지막으로 webserver.py의 SqlAlchemy 객체인코더입니다.또한 models.py 파일에 있는 메타데이터 추출기에도 의존합니다.

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

이걸로 코드골프를 해볼까 했는데

참고로 비즈니스 요건에 따라 별도로 설계된 스키마가 있기 때문에 automap_base를 사용하고 있습니다.오늘부터 SQL Chemy를 사용하기 시작했는데 문서상으로는 automap_base는 declative_base를 확장한 것으로 나타나 있습니다.이것은 SQL Chemy ORM의 전형적인 패러다임인 것 같습니다.

Tjorriemorrie 솔루션별로 외부 키를 따르는 것은 아니지만 단순히 컬럼을 값에 일치시키고 컬럼 값을 str()-ing함으로써 Python 타입을 처리합니다.우리의 값은 Python datetime으로 구성됩니다.시간과 십진수작업을 완료하기 위한 10진수 클래스 유형 결과입니다.

지나가는 사람들에게 도움이 되길 바랍니다!

나는 이것이 꽤 오래된 게시물이라는 것을 안다.@SashaB가 준 솔루션을 사용하여 필요에 따라 수정하였습니다.

여기에 다음 사항을 추가했습니다.

1. 필드 무시 리스트: 시리얼화 중 무시되는 필드 리스트
2. 필드 바꾸기 목록: 직렬화 중에 값으로 바꿀 필드 이름을 포함하는 사전입니다.
3. 메서드가 삭제되어 BaseQuery가 시리얼화되었습니다.

제 코드는 다음과 같습니다.

def alchemy_json_encoder(revisit_self = False, fields_to_expand = [], fields_to_ignore = [], fields_to_replace = {}):
   """
   Serialize SQLAlchemy result into JSon
   :param revisit_self: True / False
   :param fields_to_expand: Fields which are to be expanded for including their children and all
   :param fields_to_ignore: Fields to be ignored while encoding
   :param fields_to_replace: Field keys to be replaced by values assigned in dictionary
   :return: Json serialized SQLAlchemy object
   """
   _visited_objs = []
   class AlchemyEncoder(json.JSONEncoder):
      def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            # don't re-visit self
            if revisit_self:
                if obj in _visited_objs:
                    return None
                _visited_objs.append(obj)

            # go through each field in this SQLalchemy class
            fields = {}
            for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata' and x not in fields_to_ignore]:
                val = obj.__getattribute__(field)
                # is this field method defination, or an SQLalchemy object
                if not hasattr(val, "__call__") and not isinstance(val, BaseQuery):
                    field_name = fields_to_replace[field] if field in fields_to_replace else field
                    # is this field another SQLalchemy object, or a list of SQLalchemy objects?
                    if isinstance(val.__class__, DeclarativeMeta) or \
                            (isinstance(val, list) and len(val) > 0 and isinstance(val[0].__class__, DeclarativeMeta)):
                        # unless we're expanding this field, stop here
                        if field not in fields_to_expand:
                            # not expanding this field: set it to None and continue
                            fields[field_name] = None
                            continue

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

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

누군가 도움이 되었으면 좋겠다!

데이터 시간 필드를 하여 유형
'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)

다음 코드는 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)

Alchemy Encoder는 훌륭하지만 Decimal 값에서 실패할 수 있습니다.10진수 문제를 해결하는 개량된 인코더를 다음에 나타냅니다.

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)

sqlalchemy를 사용하여 db I에 접속하는 경우 이는 구성이 용이한 솔루션입니다.판다를 써라.

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

(Sasha B의 훌륭한 답변에 대해 약간 수정)

이것에 의해, 특히 datetime 오브젝트가 원래의 응답으로 변환되는 문자열로 변환됩니다.None:

# Standard library imports
from datetime import datetime
import json

# 3rd party imports
from sqlalchemy.ext.declarative import DeclarativeMeta

class JsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            dict = {}

            # Remove invalid fields and just get the column attributes
            columns = [x for x in dir(obj) if not x.startswith("_") and x != "metadata"]

            for column in columns:
                value = obj.__getattribute__(column)

                try:
                    json.dumps(value)
                    dict[column] = value
                except TypeError:
                    if isinstance(value, datetime):
                        dict[column] = value.__str__()
                    else:
                        dict[column] = None
            return dict

        return json.JSONEncoder.default(self, obj)

class SqlToDict:
    def __init__(self, data) -> None:
        self.data = data

    def to_timestamp(self, date):
        if isinstance(date, datetime):
            return int(datetime.timestamp(date))
        else:
            return date

    def to_dict(self) -> List:
        arr = []
        for i in self.data:
            keys = [*i.keys()]
            values = [*i]
            values = [self.to_timestamp(d) for d in values]
            arr.append(dict(zip(keys, values)))
        return arr

예를 들어 다음과 같습니다.

SqlToDict(data).to_dict()

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

아마 이런 수업을 이용할 수 있을 것이다.

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!

모든 물체는 이 기능을 가지고 있습니다.to_dict방법

되지 않은 개체를 할 때 을 사용합니다.cursor.description을 사용하다

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)

은 거는입니다.JSONEncoder모델 열 순서를 유지하고 반복적으로 정의된 열 및 관계 필드만 유지하는 버전입니다.JSON을 사용하다

import json
from datetime import datetime
from decimal import Decimal

import arrow
from sqlalchemy.ext.declarative import DeclarativeMeta

class SQLAlchemyJSONEncoder(json.JSONEncoder):
    """
    SQLAlchemy ORM JSON Encoder
    If you have a "backref" relationship defined in your SQLAlchemy model,
    this encoder raises a ValueError to stop an infinite loop.
    """

    def default(self, obj):
        if isinstance(obj, datetime):
            return arrow.get(obj).isoformat()
        elif isinstance(obj, Decimal):
            return float(obj)
        elif isinstance(obj, set):
            return sorted(obj)
        elif isinstance(obj.__class__, DeclarativeMeta):
            for attribute, relationship in obj.__mapper__.relationships.items():
                if isinstance(relationship.__getattribute__("backref"), tuple):
                    raise ValueError(
                        f'{obj.__class__} object has a "backref" relationship '
                        "that would cause an infinite loop!"
                    )
            dictionary = {}
            column_names = [column.name for column in obj.__table__.columns]
            for key in column_names:
                value = obj.__getattribute__(key)
                if isinstance(value, datetime):
                    value = arrow.get(value).isoformat()
                elif isinstance(value, Decimal):
                    value = float(value)
                elif isinstance(value, set):
                    value = sorted(value)
                dictionary[key] = value
            for key in [
                attribute
                for attribute in dir(obj)
                if not attribute.startswith("_")
                and attribute != "metadata"
                and attribute not in column_names
            ]:
                value = obj.__getattribute__(key)
                dictionary[key] = value
            return dictionary

        return super().default(obj)

이 패키지를 정상적으로 사용하고 있습니다.https://github.com/n0nSmoker/SQLAlchemy-serializer

이 작업은 모델에서 수행할 수 있습니다.

from sqlalchemy_serializer import SerializerMixin

class SomeModel(db.Model, SerializerMixin):
    ...

완전 회귀형인_dict에 추가됩니다.

item = SomeModel.query.filter(...).one()
result = item.to_dict()

무한 재귀 방지를 위한 규칙을 만들 수 있습니다.

result = item.to_dict(rules=('-somefield', '-some_relation.nested_one.another_nested_one'))

언급URL : https://stackoverflow.com/questions/5022066/how-to-serialize-sqlalchemy-result-to-json