How to Serialize Sqlalchemy Result to Json

How to serialize SqlAlchemy result to JSON?

A flat implementation

You could use something like this:

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)

and then convert to JSON using:

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

It will ignore fields that are not encodable (set them to 'None').

It doesn't auto-expand relations (since this could lead to self-references, and loop forever).

A recursive, non-circular implementation

If, however, you'd rather loop forever, you could use:

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

And then encode objects using:

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

This would encode all children, and all their children, and all their children... Potentially encode your entire database, basically. When it reaches something its encoded before, it will encode it as 'None'.

A recursive, possibly-circular, selective implementation

Another alternative, probably better, is to be able to specify the fields you want to expand:

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

You can now call it with:

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

To only expand SQLAlchemy fields called 'parents', for example.

jsonify a SQLAlchemy result set in Flask

It seems that you actually haven't executed your query. Try following:

return jsonify(json_list = qryresult.all())

[Edit]: Problem with jsonify is, that usually the objects cannot be jsonified automatically. Even Python's datetime fails ;)

What I have done in the past, is adding an extra property (like serialize) to classes that need to be serialized.

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]

And now for views I can just do:

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

Hope this helps ;)

[Edit 2019]:
In case you have more complex objects or circular references, use a library like marshmallow).

How to create sqlalchemy to json

It looks like your LabelsData object is a SQLAlchemy model. You need to serialize it before dumping it to JSON. Here's a short example that extracts all the columns from your LabelsData object and converts the results of your query to JSON:

from json import dumps
from sqlalchemy.orm import class_mapper

def serialize(model):
"""Transforms a model into a dictionary which can be dumped to JSON."""
# first we get the names of all the columns on your model
columns = [c.key for c in class_mapper(model.__class__).columns]
# then we return their values in a dict
return dict((c, getattr(model, c)) for c in columns)

# we can then use this for your particular example
serialized_labels = [
serialize(label)
for label in session.query(LabelsData).filter(LabelsData.deleted == False)
]
your_json = dumps(serialized_labels)

How to serialize SqlAlchemy join query to JSON?

sqlalchemy put in the select part on the query whatever you pass to the query method, so if you want to get 2 classes you can do:

db.session.query(Post, Users).join(Users).filter(Post.area == id, Users.id == Post.user_id).all()

this will return an array of tuples of the Post and Users class (so you will have to change the encoder to be able to recieve tuples)

EDIT:

here is a sample to how you can change your encoder to accept tuples:

class AlchemyEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, tuple):
data = {}
for obj in o:
data.update(self.parse_sqlalchemy_object(obj))
return data
if isinstance(o.__class__, DeclarativeMeta):
return self.parse_sqlalchemy_object(o)
return json.JSONEncoder.default(self, o)

def parse_sqlalchemy_object(self, o):
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


Related Topics



Leave a reply



Submit