[python] SqlAlchemy 결과를 JSON으로 직렬화하는 방법은 무엇입니까?

PYTHON / 글쓴이

Django는 DB에서 JSON 형식으로 반환되는 ORM 모델을 자동으로 직렬화합니다.

SQLAlchemy 쿼리 결과를 JSON 형식으로 직렬화하는 방법은 무엇입니까?

시도 jsonpickle.encode했지만 쿼리 객체 자체를 인코딩합니다. 시도 json.dumps(items)했지만 반환

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

SQLAlchemy ORM 객체를 JSON / XML로 직렬화하는 것이 정말 어렵습니까? 기본 시리얼 라이저가 없습니까? 오늘날 ORM 쿼리 결과를 직렬화하는 것은 매우 일반적인 작업입니다.

내가 필요한 것은 SQLAlchemy 쿼리 결과의 JSON 또는 XML 데이터 표현을 반환하는 것입니다.

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

답변

평평한 구현

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

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’라는 SQLAlchemy 필드 만 확장합니다.

답변

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

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

그런 다음 User.as_dict() 객체를 직렬화 데 합니다.

sqlalchemy 행 객체를 python dict 로 변환 에서 설명했듯이

답변

다음과 같이 RowProxy를 dict로 변환 할 수 있습니다.

 d = dict(row.items())

그런 다음 JSON으로 직렬화하십시오 ( datetime값 과 같은 것들에 대해 인코더를 지정해야 합니다) 하나의 레코드 만 원하면 관련 레코드의 전체 계층 구조가 아니라 어렵지 않습니다.

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

답변

마시멜로를 추천합니다 . 관계 및 중첩 객체를 지원하여 모델 인스턴스를 나타내는 직렬 변환기를 만들 수 있습니다.

다음은 문서에서 잘린 예입니다. ORM 모델을 보자 Author.

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

해당 클래스의 마시맬로 스키마는 다음과 같이 구성됩니다.

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

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

… 다음과 같이 사용되었습니다.

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

… 다음과 같은 출력을 생성합니다.

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

전체 Flask-SQLAlchemy 예제를 살펴보십시오. .

marshmallow-sqlalchemy특별히 호출 된 라이브러리는 SQLAlchemy와 마시맬로를 통합합니다. 해당 라이브러리에서 Author위에서 설명한 모델 의 스키마는 다음과 같습니다.

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

통합을 통해 SQLAlchemy Column유형 에서 필드 유형을 유추 할 수 있습니다.

마시멜로 -sqlalchemy.

답변

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

답변

Flask-JsonTools 패키지에는 JsonSerializableBase 구현이 있습니다. 모델에 대한 Base 클래스 되어 있습니다.

용법:

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

Base = declarative_base(cls=(JsonSerializableBase,))

class User(Base):
    #...

이제 User 모델은 마술처럼 직렬화 가능합니다.

프레임 워크가 Flask가 아닌 경우 코드를 잡을 수 있습니다.

답변

보안상의 이유로 모든 모델 필드를 반환해서는 안됩니다. 선택적으로 선택하는 것을 선호합니다.

이제 인코딩 플라스크의 JSON은 UUID, 날짜 및 관계 (및 추가 지원 query과 query_class에 대한 flask_sqlalchemy의 db.Model클래스). 인코더를 다음과 같이 업데이트했습니다.

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

이를 통해 __json__인코딩하려는 필드 목록을 반환 하는 속성을 선택적으로 추가 할 수 있습니다 .

app/models.py

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

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

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

내 뷰에 @jsonapi를 추가하고 결과 목록을 반환하면 출력은 다음과 같습니다.

[

{

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

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

]