SQLAlchemy JSON serialization

python flask sqlalchemy

Although, in these times, this should be a very trivial matter, I found some issues along the internet related to serialize SQLAlchemy model objects to JSON. Here's the way I do it.

Create a dictionary

The first thing to do is create a dictionary from the model:

{column.key: getattr(self, attr) for attr, column in self.__mapper__.c.items()}

Accessing to __mapper__.columns keys instead of __table__.columns we'll avoid problems with attributes that don't match the column name, being able to get the attribute name and the column.name. For example, according to naming conventions, in Python the variable/attribute names must be underscored and in JSON, the keys must be camelCased:

author_id = db.Column('authorId', db.Integer, db.ForeignKey('users.id'))

This field must be expressed in JSON with the column name given, authorId instead of author_id.

Relationships

In the same way, we can access to the relationships from __mapper__.relationships:

relationships = {}
for attr, relation in self.__mapper__.relationships.items():
    value = getattr(self, attr)
    if value is None:
        relationships[relation.key] = None
    elif isinstance(value.__class__, DeclarativeMeta):
        relationships[relation.key] = value.to_dict()
    else:
        relationships[relation.key] = [i.to_dict() for i in value]

Accessing to multiple relationships may be a hard SQL query and it may end in an infinite recursive loop and causing a RuntimeError. We will avoid this between two tables, but more than two tables may be involved, depending on the data modelling. We will need to use the relationships serialization carefully.

The JSON encoder

We can write our JSON encoder as complex as we need it. In this example, our JSON encoder will support UUID and datetime instances:

def to_json(self, rel=None):
    def extended_encoder(x):
        if isinstance(x, datetime):
            return x.isoformat()
        if isinstance(x, UUID):
            return str(x)
    return json.dumps(self.to_dict(), default=extended_encoder)

Flask support

We won't need the previous step if we are working with Flask, since the JSON encoder will be applied to the model instance by the jsonify function. We will need to define the __iter__ magic method in the model because dict type will be applied in the jsonify function:

def __iter__(self):
    return self.to_dict().iteritems()

Also, we will need to set our JSON encoder to support SQLAlchemy model instances:

from sqlalchemy.ext.declarative import DeclarativeMeta
from flask import Flask
from flask.json import JSONEncoder


class CustomJSONEncoder(JSONEncoder):

    def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            return obj.to_dict()
        return super(CustomJSONEncoder, self).default(obj)


app = Flask(__name__)
app.json_encoder = CustomJSONEncoder

Putting all together

We can define a mixin to use along with our models like:

import json
from uuid import UUID

from sqlalchemy.ext.declarative import DeclarativeMeta


class OutputMixin(object):
    RELATIONSHIPS_TO_DICT = False

    def __iter__(self):
        return self.to_dict().iteritems()

    def to_dict(self, rel=None, backref=None):
        if rel is None:
            rel = self.RELATIONSHIPS_TO_DICT
        res = {column.key: getattr(self, attr)
               for attr, column in self.__mapper__.c.items()}
        if rel:
            for attr, relation in self.__mapper__.relationships.items():
                # Avoid recursive loop between to tables.
                if backref == relation.table:
                    continue
                value = getattr(self, attr)
                if value is None:
                    res[relation.key] = None
                elif isinstance(value.__class__, DeclarativeMeta):
                    res[relation.key] = value.to_dict(backref=self.__table__)
                else:
                    res[relation.key] = [i.to_dict(backref=self.__table__)
                                         for i in value]
        return res

    def to_json(self, rel=None):
        def extended_encoder(x):
            if isinstance(x, datetime):
                return x.isoformat()
            if isinstance(x, UUID):
                return str(x)
        if rel is None:
            rel = self.RELATIONSHIPS_TO_DICT
        return json.dumps(self.to_dict(rel), default=extended_encoder)

A simple example

Here's a simple example of models for a blog:

from datetime import datetime
import json
from uuid import uuid4

from sqlalchemy.dialects import postgres

from app import db


tags_association = db.Table(
    'tags_association',
    db.Column('tag_id', db.Integer, db.ForeignKey('tags.id')),
    db.Column('entry_id', db.Integer, db.ForeignKey('entries.id')))


class Entry(OutputMixin, db.Model):
    __tablename__ = 'entries'
    RELATIONSHIPS_TO_DICT = True

    id = db.Column(db.Integer, primary_key=True)
    uuid = db.Column(postgres.UUID(as_uuid=True), unique=True, default=uuid4)
    title = db.Column(db.String(80), nullable=False)
    body = db.Column(db.String(), nullable=False)
    created = db.Column(db.DateTime, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow)
    author_id = db.Column('authorId', db.Integer, db.ForeignKey('users.id'))
    author = db.relationship('User',
                             backref=db.backref('entries', lazy='dynamic'))
    tags = db.relationship('Tag',
                           secondary=tags_association,
                           backref=db.backref('entries', lazy='dynamic'))


class Tag(OutputMixin, db.Model):
    __tablename__ = 'tags'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(20), nullable=False)


class User(OutputMixin, db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(80), unique=True, nullable=False)
from uuid import UUID
from app.models import Entry

uuid = UUID('ae89ee70-a2b0-49cf-b2d7-f1dbc1c6827e')
entry = Entry.query.filter_by(uuid=uuid).first_or_404()
entry.to_json()
{
  "author": {
    "email": "mmast@gmx.com", 
    "id": 1, 
    "username": "modesto"
  }, 
  "authorId": 1, 
  "body": "the body", 
  "created": "Sat, 21 Nov 2015 18:20:12 GMT", 
  "id": 1, 
  "tags": [
    {
      "id": 1, 
      "name": "tag1"
    }, 
    {
      "id": 2, 
      "name": "tag2"
    }
  ], 
  "title": "the title", 
  "updated": "Thu, 26 Nov 2015 15:04:20 GMT", 
  "uuid": "ae89ee70-a2b0-49cf-b2d7-f1dbc1c6827e"
}