使用原始 SQL 时在 SQLAlchemy 中反序列化 JSON [英] Deserializing JSON in SQLAlchemy when using raw SQL

问题描述

我有一个存储在文本列中的 JSON 表:

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)

我在 SQLAlchemy 中定义了到这个表的映射，使用在编组 JSON 字符串":

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)

如果我查询所有 T，attrs 会从 JSON 中反序列化:

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)

但是如果我使用文本查询/原始 SQL，它不会被反序列化:

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}'

使用原始 SQL 查询时，如何让 SQLAlchemy 反序列化 attrs 列?

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

行为与其他数据库(MySQL、Postgres)相同.我使用的数据库 (MySQL 5.5) 不支持原生 JSON 类型，因此无法更改列类型.

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.

推荐答案

你可以告诉 TextClause(由 text()) 使用 .columns():

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)

或者，对于 SQLAlchemy<0.9，使用 typemap 参数:

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)

这篇关于使用原始 SQL 时在 SQLAlchemy 中反序列化 JSON的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！