Flask에서 SQLAlchemy 결과 세트를 jsonify

Flask에서 SQLAlchemy 결과 세트를 jsonify

---

이 질문에는 이미 답변이 있습니다.

• SqlAlchemy 결과를 JSON으로 직렬화하는 방법은 무엇입니까? 답변 21 개

Flask / Python에서 SQLAlchemy 결과 세트를 jsonify하려고합니다.

플라스크 메일 링리스트는 다음과 같은 방법을 제안 http://librelist.com/browser//flask/2011/2/16/jsonify-sqlalchemy-pagination-collection-result/#04a0754b63387f87e59dda564bde426e를 :

return jsonify(json_list = qryresult)

그러나 다음과 같은 오류가 발생합니다.

TypeError: <flaskext.sqlalchemy.BaseQuery object at 0x102c2df90> 
is not JSON serializable

여기서 무엇을 간과합니까?

이 질문을 발견했습니다 : SqlAlchemy 결과를 JSON으로 직렬화하는 방법은 무엇입니까? 메일 링리스트 포스트가 제안한 것처럼 Flask가 더 쉬운 마법을 가지고 있는지 알지 못했습니다.

편집 : 명확하게하기 위해 이것이 내 모델의 모습입니다.

class Rating(db.Model):

    __tablename__ = 'rating'

    id = db.Column(db.Integer, primary_key=True)
    fullurl = db.Column(db.String())
    url = db.Column(db.String())
    comments = db.Column(db.Text)
    overall = db.Column(db.Integer)
    shipping = db.Column(db.Integer)
    cost = db.Column(db.Integer)
    honesty = db.Column(db.Integer)
    communication = db.Column(db.Integer)
    name = db.Column(db.String())
    ipaddr = db.Column(db.String())
    date = db.Column(db.String())

    def __init__(self, fullurl, url, comments, overall, shipping, cost, honesty, communication, name, ipaddr, date):
        self.fullurl = fullurl
        self.url = url
        self.comments = comments
        self.overall = overall
        self.shipping = shipping
        self.cost = cost
        self.honesty = honesty
        self.communication = communication
        self.name = name
        self.ipaddr = ipaddr
        self.date = date

---

실제로 쿼리를 실행하지 않은 것 같습니다. 다음을 시도하십시오.

return jsonify(json_list = qryresult.all())

[편집] : jsonify의 문제점은 일반적으로 객체가 자동으로 jsonified 될 수 없다는 것입니다. 파이썬의 날짜 시간조차도 실패합니다.)

내가 과거에 한 일은 serialize직렬화 해야하는 클래스에 추가 속성 (예 :)을 추가하는 것입니다.

def dump_datetime(value):
    """Deserialize datetime object into string form for JSON processing."""
    if value is None:
        return None
    return [value.strftime("%Y-%m-%d"), value.strftime("%H:%M:%S")]

class Foo(db.Model):
    # ... SQLAlchemy defs here..
    def __init__(self, ...):
       # self.foo = ...
       pass

    @property
    def serialize(self):
       """Return object data in easily serializable format"""
       return {
           'id'         : self.id,
           'modified_at': dump_datetime(self.modified_at),
           # This is an example how to deal with Many2Many relations
           'many2many'  : self.serialize_many2many
       }
    @property
    def serialize_many2many(self):
       """
       Return object's relations in easily serializable format.
       NB! Calls many2many's serialize property.
       """
       return [ item.serialize for item in self.many2many]

그리고 지금보기를 위해 나는 단지 할 수 있습니다 :

return jsonify(json_list=[i.serialize for i in qryresult.all()])

도움이 되었기를 바랍니다 ;)

[Edit 2019] : 더 복잡한 객체 또는 순환 참조가있는 경우 marshmallow 와 같은 라이브러리를 사용하십시오 .

---

json으로 직렬화하기 위해 동일한 요구가있었습니다. 이 질문을 살펴보십시오 . 프로그래밍 방식으로 열을 검색하는 방법을 보여줍니다. 그래서 그로부터 아래 코드를 만들었습니다. 그것은 나를 위해 작동하며 내 웹 응용 프로그램에서 사용할 것입니다. 행복한 코딩!

def to_json(inst, cls):
    """
    Jsonify the sql alchemy query result.
    """
    convert = dict()
    # add your coversions for things like datetime's 
    # and what-not that aren't serializable.
    d = dict()
    for c in cls.__table__.columns:
        v = getattr(inst, c.name)
        if c.type in convert.keys() and v is not None:
            try:
                d[c.name] = convert[c.type](v)
            except:
                d[c.name] = "Error:  Failed to covert using ", str(convert[c.type])
        elif v is None:
            d[c.name] = str()
        else:
            d[c.name] = v
    return json.dumps(d)

class Person(base):
    __tablename__ = 'person'
    id = Column(Integer, Sequence('person_id_seq'), primary_key=True)
    first_name = Column(Text)
    last_name = Column(Text)
    email = Column(Text)

    @property
    def json(self):
        return to_json(self, self.__class__)

---

일반적으로 나에게 충분한 것은 다음과 같습니다.

모델과 함께 사용하는 직렬화 믹스 인을 만듭니다. 직렬화 함수는 기본적으로 SQLAlchemy 인스펙터가 노출하는 모든 속성을 가져와이를 dict에 넣습니다.

from sqlalchemy.inspection import inspect

class Serializer(object):

    def serialize(self):
        return {c: getattr(self, c) for c in inspect(self).attrs.keys()}

    @staticmethod
    def serialize_list(l):
        return [m.serialize() for m in l]

이제 필요한 것은 Serializermixin 클래스 를 사용하여 SQLAlchemy 모델을 확장하는 것 입니다.

노출하지 않거나 특수한 형식이 필요한 필드가 serialize()있는 경우 모델 서브 클래스 의 함수를 대체하십시오 .

class User(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String)
    password = db.Column(db.String)

    # ...

    def serialize(self):
        d = Serializer.serialize(self)
        del d['password']
        return d

컨트롤러에서 수행해야 할 일은 결과에서 serialize()함수 를 호출하는 것입니다 (또는 serialize_list(l)쿼리 결과가 목록에있는 경우).

def get_user(id):
    user = User.query.get(id)
    return json.dumps(user.serialize())

def get_users():
    users = User.query.all()
    return json.dumps(User.serialize_list(users))

---

여기 내 접근 방식이 있습니다 : https://github.com/n0nSmoker/SQLAlchemy-serializer

pip install SQLAlchemy- 시리얼 라이저

모델에 믹스 인을 쉽게 추가 할 수 있으며 인스턴스에서 .to_dict () 메소드를 호출하는 것보다

SerializerMixin 기반으로 자신의 믹스 인을 작성할 수도 있습니다.

---

단순 쿼리 (조인 없음)의 경우 다음을 수행 할 수 있습니다.

@app.route('/results/')
def results():
    data = Table.query.all()
    result = [d.__dict__ for d in data]
    return jsonify(result=result)

데이터베이스에서 특정 열만 반환하려면이 작업을 수행 할 수 있습니다

@app.route('/results/')
def results():
    cols = ['id', 'url', 'shipping']
    data = Table.query.all()
    result = [{col: getattr(d, col) for col in cols} for d in data]
    return jsonify(result=result)

---

좋아, 나는 몇 시간 동안이 일을 해 왔으며, 가장 파이썬적인 솔루션이라고 생각하는 것을 개발했습니다. 다음 코드 스 니펫은 python3이지만 필요한 경우 백 포트하기에 너무 심하게 고통스럽지 않아야합니다.

우리가 할 첫 번째 일은 db 모델을 dicts 처럼 작동하게하는 mixin으로 시작하는 것입니다 .

from sqlalchemy.inspection import inspect

class ModelMixin:
    """Provide dict-like interface to db.Model subclasses."""

    def __getitem__(self, key):
        """Expose object attributes like dict values."""
        return getattr(self, key)

    def keys(self):
        """Identify what db columns we have."""
        return inspect(self).attrs.keys()

이제 우리는 모델을 정의하고 믹스 인을 상속합니다 :

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)
    # etc ...

즉, 인스턴스 통과 할 수있을하는 데 걸리는 전부 MyModel()에를 dict()실제 라이브 얻을 dict우리가 만드는쪽으로 꽤 먼 길을 얻을 수있는, 그것의 예를 jsonify()이해합니다. 다음으로 우리 JSONEncoder는 나머지 길을 갖도록 확장해야합니다 .

from flask.json import JSONEncoder
from contextlib import suppress

class MyJSONEncoder(JSONEncoder):
    def default(self, obj):
        # Optional: convert datetime objects to ISO format
        with suppress(AttributeError):
            return obj.isoformat()
        return dict(obj)

app.json_encoder = MyJSONEncoder

보너스 포인트 : 모델에 계산 필드가 포함 된 경우 (즉, JSON 출력에 실제로 데이터베이스에 저장되지 않은 필드가 포함되도록하려는 경우) 매우 쉽습니다. 계산 된 필드를 @propertys 로 정의하고 다음 keys()과 같이 메소드를 확장하십시오 .

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)

    @property
    def computed_field(self):
        return 'this value did not come from the db'

    def keys(self):
        return super().keys() + ['computed_field']

이제 jsonify는 사소한 일입니다.

@app.route('/whatever', methods=['GET'])
def whatever():
    return jsonify(dict(results=MyModel.query.all()))

---

당신이 사용하는 경우 flask-restful당신이 사용할 수있는 원수를 :

from flask.ext.restful import Resource, fields, marshal

topic_fields = {
    'title':   fields.String,
    'content': fields.String,
    'uri':     fields.Url('topic'),
    'creator': fields.String,
    'created': fields.DateTime(dt_format='rfc822')
}

class TopicListApi(Resource):
    def get(self):
        return {'topics': [marshal(topic, topic_fields) for topic in DbTopic.query.all()]}

반환하는 내용과 유형을 명시 적으로 나열해야합니다. 어쨌든 API에 선호합니다. 직렬화는 쉽게 처리 할 수 ​​있으며 (필요 없음 jsonify) 날짜도 문제가되지 않습니다. uri필드 의 컨텐츠 는 topic엔드 포인트 및 ID를 기반으로 자동 생성됩니다 .

---

나는이 문제를 하루 동안 더 잘 보았고 여기에 내가 생각해 낸 것이 있습니다 ( 이 방향으로 나를 가리 키기 위해 https://stackoverflow.com/a/5249214/196358 에 신용 ).

(참고 : flask-sqlalchemy를 사용하고 있으므로 모델 선언 형식이 sqlalchemy와 약간 다릅니다.)

내 models.py파일에서 :

import json

class Serializer(object):
  __public__ = None
  "Must be implemented by implementors"

  def to_serializable_dict(self):
    dict = {}
    for public_key in self.__public__:
      value = getattr(self, public_key)
      if value:
        dict[public_key] = value
    return dict

class SWEncoder(json.JSONEncoder):
  def default(self, obj):
    if isinstance(obj, Serializer):
      return obj.to_serializable_dict()
    if isinstance(obj, (datetime)):
      return obj.isoformat()
    return json.JSONEncoder.default(self, obj)


def SWJsonify(*args, **kwargs):
  return current_app.response_class(json.dumps(dict(*args, **kwargs), cls=SWEncoder, indent=None if request.is_xhr else 2), mimetype='application/json')
  # stolen from https://github.com/mitsuhiko/flask/blob/master/flask/helpers.py

내 모든 모델 객체는 다음과 같습니다.

class User(db.Model, Serializer):
  __public__ = ['id','username']
  ... field definitions ...

내 관점에서, 나는 전화했을 때마다 Jsonify다음과 같이 SWJsonify를 호출합니다 .

@app.route('/posts')
def posts():
  posts = Post.query.limit(PER_PAGE).all()
  return SWJsonify({'posts':posts })

꽤 잘 작동하는 것 같습니다. 심지어 관계에서도. YMMV와는 거리가 멀지 만 지금까지는 "정확한"느낌입니다.

제안을 환영합니다.

---

다음은 모든 클래스에 as_dict () 메소드를 추가하는 방법과 모든 단일 클래스에 갖고 싶은 다른 메소드입니다. 이것이 원하는 방법인지 아닌지 확실하지 않지만 작동합니다 ...

class Base(object):
    def as_dict(self):
        return dict((c.name,
                     getattr(self, c.name))
                     for c in self.__table__.columns)


Base = declarative_base(cls=Base)

---

ActiveRecord to_json에서 사용 된 레일 접근 방식과 같은 것을 찾고 있었고 다른 제안에 만족하지 못하고이 Mixin을 사용하여 비슷한 것을 구현했습니다. 중첩 된 모델을 처리하고 최상위 또는 중첩 된 모델의 속성을 포함하거나 제외합니다.

class Serializer(object):

    def serialize(self, include={}, exclude=[], only=[]):
        serialized = {}
        for key in inspect(self).attrs.keys():
            to_be_serialized = True
            value = getattr(self, key)
            if key in exclude or (only and key not in only):
                to_be_serialized = False
            elif isinstance(value, BaseQuery):
                to_be_serialized = False
                if key in include:
                    to_be_serialized = True
                    nested_params = include.get(key, {})
                    value = [i.serialize(**nested_params) for i in value]

            if to_be_serialized:
                serialized[key] = value

        return serialized

그런 다음 BaseQuery 직렬화 가능을 얻기 위해 BaseQuery를 확장했습니다.

class SerializableBaseQuery(BaseQuery):

    def serialize(self, include={}, exclude=[], only=[]):
        return [m.serialize(include, exclude, only) for m in self]

다음 모델의 경우

class ContactInfo(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    full_name = db.Column(db.String())
    source = db.Column(db.String())
    source_id = db.Column(db.String())

    email_addresses = db.relationship('EmailAddress', backref='contact_info', lazy='dynamic')
    phone_numbers = db.relationship('PhoneNumber', backref='contact_info', lazy='dynamic')


class EmailAddress(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    email_address = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey('contact_info.id'))


class PhoneNumber(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    phone_number = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey('contact_info.id'))

    phone_numbers = db.relationship('Invite', backref='phone_number', lazy='dynamic')

당신은 같은 것을 할 수 있습니다

@app.route("/contact/search", methods=['GET'])
def contact_search():
    contact_name = request.args.get("name")
    matching_contacts = ContactInfo.query.filter(ContactInfo.full_name.like("%{}%".format(contact_name)))

    serialized_contact_info = matching_contacts.serialize(
        include={
            "phone_numbers" : {
                "exclude" : ["contact_info", "contact_info_id"]
            },
            "email_addresses" : {
                "exclude" : ["contact_info", "contact_info_id"]
            }
        }
    )

    return jsonify(serialized_contact_info)

---

선언적 기반을 사용하는 경우 내 답변은 다음과 같습니다 (이미 게시 된 답변 중 일부의 도움으로).

# in your models definition where you define and extend declarative_base()
from sqlalchemy.ext.declarative import declarative_base
...
Base = declarative_base()
Base.query = db_session.query_property()
...

# define a new class (call "Model" or whatever) with an as_dict() method defined
class Model():
    def as_dict(self):
        return { c.name: getattr(self, c.name) for c in self.__table__.columns }

# and extend both the Base and Model class in your model definition, e.g.
class Rating(Base, Model):
    ____tablename__ = 'rating'
    id = db.Column(db.Integer, primary_key=True)
    fullurl = db.Column(db.String())
    url = db.Column(db.String())
    comments = db.Column(db.Text)
    ...

# then after you query and have a resultset (rs) of ratings
rs = Rating.query.all()

# you can jsonify it with
s = json.dumps([r.as_dict() for r in rs], default=alchemyencoder)
print (s)

# or if you have a single row
r = Rating.query.first()

# you can jsonify it with
s = json.dumps(r.as_dict(), default=alchemyencoder)

# you will need this alchemyencoder where your are calling json.dumps to handle datetime and decimal format
# credit to Joonas @ http://codeandlife.com/2014/12/07/sqlalchemy-results-to-json-the-easy-way/
def alchemyencoder(obj):
    """JSON encoder function for SQLAlchemy special classes."""
    if isinstance(obj, datetime.date):
        return obj.isoformat()
    elif isinstance(obj, decimal.Decimal):
        return float(obj)

---

Flask-Restful 0.3.6 요청 파싱 권장 마시멜로

marshmallow는 ORM / ODM / framework-agnostic 라이브러리로, 객체와 같은 복잡한 데이터 유형을 원시 Python 데이터 유형으로 변환합니다.

아래는 간단한 마시멜로 예제입니다.

from marshmallow import Schema, fields

class UserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    created_at = fields.DateTime()

from marshmallow import pprint

user = User(name="Monty", email="monty@python.org")
schema = UserSchema()
result = schema.dump(user)
pprint(result)
# {"name": "Monty",
#  "email": "monty@python.org",
#  "created_at": "2014-08-17T14:54:16.049594+00:00"}

핵심 기능은

스키마 선언
객체 직렬화 ( "덤핑")
객체 직렬화 해제 ( "로드")
객체 컬렉션 처리
유효성 검사
속성 이름
지정 직렬화 / 역 직렬화 키 지정
리팩토링 : 암시 적 필드 생성
순서 출력
"읽기 전용"및 "쓰기 전용"필드
기본값 지정 직렬화 / 직렬화 값
중첩 스키마
사용자 정의 필드

---

jobDict라는 RowProxy 객체 목록의 SQL 쿼리 defaultdict를 사용하여 작업했습니다. 객체 유형이 무엇인지 알아내는 데 시간이 걸렸습니다.

이것은 행을 목록으로 형변환하고 처음에 값을 사용하여 dict를 정의하여 깨끗한 jsonEncoding으로 해결할 수있는 매우 간단한 빠른 방법이었습니다.

    jobDict = defaultdict(list)
    def set_default(obj):
        # trickyness needed here via import to know type
        if isinstance(obj, RowProxy):
            return list(obj)
        raise TypeError


    jsonEncoded = json.dumps(jobDict, default=set_default)

---

방금 방법을 추가하고 싶습니다.

db 모델을 처리하기 위해 custome json 인코더를 정의하십시오.

class ParentEncoder(json.JSONEncoder):
    def default(self, obj):
        # convert object to a dict
        d = {}
        if isinstance(obj, Parent):
            return {"id": obj.id, "name": obj.name, 'children': list(obj.child)}
        if isinstance(obj, Child):
            return {"id": obj.id, "name": obj.name}

        d.update(obj.__dict__)
        return d

그런 다음보기 기능에서

parents = Parent.query.all()
dat = json.dumps({"data": parents}, cls=ParentEncoder)
resp = Response(response=dat, status=200, mimetype="application/json")
return (resp)

부모가 관계가 있지만 잘 작동합니다.

---

많은 시간이 걸리고 유효한 답변이 많이 있지만 다음 코드 블록이 작동하는 것 같습니다.

my_object = SqlAlchemyModel()
my_serializable_obj = my_object.__dict__
del my_serializable_obj["_sa_instance_state"]
print(jsonify(my_serializable_object))

나는 이것이 완벽한 해결책이 아니며 다른 것만 큼 우아하지 않다는 것을 알고 있지만 빠른 수정을 원하는 사람들은 이것을 시도 할 수 있습니다.

참고 URL : https://stackoverflow.com/questions/7102754/jsonify-a-sqlalchemy-result-set-in-flask