python - 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.

Answer 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)