Loading SQLAlchemy Models from JSON

DONG Yuxuan @ Apr 11, 2019


Abstract

Convert a JSON-encodable dictionary to an SQLAlchemy model


I recently wrote an article about dumping SQLAlchemy models. The inverse problem, loading models from JSON, is more complex. The main problem is about dealing with relationships. While loading a model, if relationships are dumped using the technology introduced by my recent article, which means only primary keys of remote sides of the relationship are dumped, it will be simple because we can just fetch remote sides from the database using their primary keys and populate the associated field with them. However, the situation of loading from nested JSON happens often. For example, considering a school management system, the following class may be submitted to create:

{
	"name": "Super class",
	"grade": 2,
	"students": [
		{"name": "John Smith", "age": 8, "gender": 1},
		{"name": "Jack Lee", "age": 7, "gender": 1},
		{"name": "Rose Obama", "age": 7, "gender": 0}
	]
}

In this example, the class and all students of the class are new. None of them is currently in the databse thus there is no possibily they’re dumped as primary keys. The example can be modified to a more complicated one which is the hybrid of primary keys and nested objects:

{
	"name": "Super class",
	"grade": 2,
	"students": [
		{"name": "John Smith", "age": 8, "gender": 1},
		3671,
		{"name": "Rose Obama", "age": 7, "gender": 0}
	]
}

The student named Jack Lee is replaced with an existed student whose ID the primary key is 3671. Once these situations are analyzed carefully, a model loader can be written. For simplicity, we assume the model has one and only one column as the primary key.

''' Usage
model = dict2sqla(the_dict_from_json, cls=the_model_class, session=sqlalchemy_session)
'''

from sqlalchemy.orm.properties import ColumnProperty
from sqlalchemy.orm.relationships import RelationshipProperty

def _get_remote(remote, remotecls, session):
	if isinstance(remote, dict):
		return dict2sqla(remote, remotecls, session)
	else:
		return session.query(remotecls).get(remote)

def dict2sqla(dct, cls, session, ignores=None):
	ret = cls()
	ignores = ignores or hasattr(cls, "__ignores__") and cls.__ignores__ or []

	for prop in cls.__mapper__.iterate_properties:
		key = prop.key
		if key in ignores or key not in dct:
			continue
		val = dct[key]

		if isinstance(prop, ColumnProperty):
			setattr(ret, key, val)

		elif isinstance(prop, RelationshipProperty):
			remotecls = getattr(cls, key).property.mapper.class_

			if isinstance(val, list):
				setattr(ret, key, [_get_remote(v, remotecls, session) for v in val])
			else:
				setattr(ret, key, _get_remote(val, remotecls, session))

	return ret

The code can also be found at Gist.