Deserializing JSON in SQLAlchemy when using raw SQL

Question

I have a table with JSON stored in a text column:

import json
from sqlalchemy import create_engine, Column, text, Integer, TEXT, TypeDecorator
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///:memory:')
engine.execute("create table t (t_id int not null primary key, attrs text not null)")
engine.execute("insert into t values (1, '{\"a\": 1, \"b\": 2}')")

Session = sessionmaker(bind=engine)

I defined a mapping to this table in SQLAlchemy, using the custom type defined in the SQLAlchemy docs under "Marshal JSON Strings":

Base = declarative_base()

# http://docs.sqlalchemy.org/en/rel_1_1/core/custom_types.html#marshal-json-strings
class JSONEncodedDict(TypeDecorator):
    impl = TEXT

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = json.dumps(value)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = json.loads(value)
        return value

class T(Base):
    __tablename__ = 't'

    t_id = Column(Integer, primary_key=True)
    attrs = Column(JSONEncodedDict)

If I query all Ts, attrs gets deserialized from JSON:

session = Session()
t = session.query(T).first()
assert type(t.attrs) == dict, repr(t.attrs)

But if I use a textual query / raw SQL, it isn't deserialized:

session = Session()
t = session.query(T).from_statement(text('select * from t')).first()
assert type(t.attrs) == dict, repr(t.attrs)  # AssertionError: u'{"a": 1, "b": 2}'

How do I make SQLAlchemy deserialize the attrs column when querying with raw SQL?

The behavior is the same with other databases (MySQL, Postgres). The database I am using (MySQL 5.5) does not support native JSON types, so changing the column type is not an option.

Ilja Everilä · Accepted Answer · 2020-04-24 05:46:04Z

1

You can tell TextClause (produced by text()) the column types using .columns():

from sqlalchemy import inspect

session = Session()
stmt = text('select * from t').columns(*inspect(T).columns)
t = session.query(T).from_statement(stmt).first()
assert type(t.attrs) == dict, repr(t.attrs)

Or, for SQLAlchemy<0.9, use the typemap argument:

from sqlalchemy import inspect

session = Session()
typemap = {c.name: c.type for c in inspect(T).columns}
stmt = text('select * from t', typemap=typemap)
t = session.query(T).from_statement(stmt).first()
assert type(t.attrs) == dict, repr(t.attrs)

edited Apr 24, 2020 at 5:46

Ilja Everilä

53.4k9 gold badges138 silver badges142 bronze badges

answered Nov 2, 2017 at 21:04

Snowball

11.8k3 gold badges38 silver badges51 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Ilja Everilä Over a year ago

columns() also accepts keyword arguments, allowing a "shortcut" form for defining types of columns, for example stmt.columns(attrs=JSONEncodedDict) in this case.

Collectives™ on Stack Overflow

Deserializing JSON in SQLAlchemy when using raw SQL

1 Answer 1

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related