升级SQLAlchemy 1.3至1.4后Hybrid Property失效，返回anon_1而非列名

升级SQLAlchemy到1.4版本后，使用hybrid_property映射JSONB属性的功能失效。执行查询时，字典形式的结果无法获取正确列名，返回{'anon_1': 1881}而非预期的{'born': 1881}。相关代码如下：

import pytest
from sqlalchemy import Column, Integer, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.dialects.postgresql import JSONB

import os

from sqlalchemy import create_engine
from sqlalchemy.event import listens_for
from sqlalchemy.pool import StaticPool
from sqlalchemy.orm import Session

def create_postgres():
    engine = create_engine(os.environ["TEST_DATABASE"], poolclass=StaticPool)
    engine.execute("SET search_path TO pg_temp")

    @listens_for(engine, "engine_connect")
    def set_search_path(connection, branch):
        connection.execute("SET search_path TO pg_temp")

    return engine

@pytest.fixture
def session(request):
    engine = create_postgres()

    Base.metadata.create_all(engine)
    return Session(engine)


def update_json_property_names(attrs):
    for key, value in attrs.items():
        if isinstance(value, hybrid_property) and hasattr(value.fget, "__json_property"):
            setattr(value.fget, "prop_key", key)

def json_property(json_column, name, type_=None):
    def instance_get(self):
        return getattr(self, json_column.key)[name]

    def instance_set(self, value):
        json_obj = getattr(self, json_column.key)
        if json_obj is None:
            setattr(self, json_column.key, {})
        getattr(self, json_column.key)[name] = value

    def cls_get(cls):
        if type_:
            return json_column.cast(JSONB)[name].astext.cast(type_)
        else:
            return json_column.cast(JSONB)[name]

    prop = hybrid_property(
        fget=instance_get,
        expr=cls_get,
        fset=instance_set
    )

    prop.__json_property = (json_column.key, name)

    return prop


class ModelMetaType(DeclarativeMeta):
    def __new__(cls, name, bases, attrs):
        update_json_property_names(attrs)
        return DeclarativeMeta.__new__(cls, name, bases, attrs)



Base = declarative_base(metaclass=ModelMetaType)


class Author(Base):
    __tablename__ = "author"

    id = Column(Integer, primary_key=True)
    data = Column(JSONB, nullable=False)

    born = json_property(data, "yearBorn", Integer())
    died = json_property(data, "yearDied")

    def __repr__(self):
        return "Author(id={!r}, name={!r}, data={!r})".format(
            self.id, self.name, self.data
        )

def test_expression_has_name_of_json_property(session):
    session.add(Author(data={"yearBorn": 1881}))
    session.commit()
    q = select(Author.born)
    result = session.execute(q).one()
    assert result._asdict() == {"born": 1881}

需要调整json_property的类级表达式生成逻辑，为查询结果指定明确的列别名，同时修改元类的属性绑定逻辑，确保别名与属性名一致。具体修改如下：

1. 修改json_property函数，为表达式添加别名 #

SQLAlchemy 1.4会为无别名的匿名表达式自动生成anon_*名称，因此需要在类级表达式中显式指定属性名作为别名。同时简化冗余的JSONB类型转换：

def json_property(json_column, name, type_=None):
    prop_key = None  # 后续由元类注入属性名

    def instance_get(self):
        return getattr(self, json_column.key)[name]

    def instance_set(self, value):
        json_obj = getattr(self, json_column.key)
        if json_obj is None:
            setattr(self, json_column.key, {})
        getattr(self, json_column.key)[name] = value

    def cls_get(cls):
        expr = json_column[name]  # 原json_column已是JSONB类型，无需重复cast
        if type_:
            expr = expr.astext.cast(type_)
        return expr.label(prop_key)  # 使用属性名作为列别名

    prop = hybrid_property(
        fget=instance_get,
        expr=cls_get,
        fset=instance_set
    )

    prop.__json_property = (json_column.key, name)
    prop.cls_get = cls_get  # 暴露cls_get，方便元类设置prop_key
    prop.cls_get.prop_key = prop_key

    return prop

2. 修改update_json_property_names函数，绑定属性名到表达式 #

在元类处理时，将属性名（如born）赋值给cls_get的prop_key变量，确保表达式生成时使用正确的别名：

def update_json_property_names(attrs):
    for key, value in attrs.items():
        if isinstance(value, hybrid_property) and hasattr(value, "__json_property"):
            value.cls_get.prop_key = key  # 将属性名注入到cls_get的prop_key中

3. 修正Author类的__repr__方法（可选） #

原代码中__repr__引用了未定义的self.name，可以修正为：

def __repr__(self):
    return "Author(id={!r}, data={!r})".format(
        self.id, self.data
    )

修改后的完整测试代码 #

import pytest
from sqlalchemy import Column, Integer, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta
from sqlalchemy.ext.hybrid import hybrid_property

import os

from sqlalchemy import create_engine
from sqlalchemy.event import listens_for
from sqlalchemy.pool import StaticPool
from sqlalchemy.orm import Session

def create_postgres():
    engine = create_engine(os.environ["TEST_DATABASE"], poolclass=StaticPool)
    engine.execute("SET search_path TO pg_temp")

    @listens_for(engine, "engine_connect")
    def set_search_path(connection, branch):
        connection.execute("SET search_path TO pg_temp")

    return engine

@pytest.fixture
def session(request):
    engine = create_postgres()

    Base.metadata.create_all(engine)
    return Session(engine)


def update_json_property_names(attrs):
    for key, value in attrs.items():
        if isinstance(value, hybrid_property) and hasattr(value, "__json_property"):
            value.cls_get.prop_key = key

def json_property(json_column, name, type_=None):
    prop_key = None

    def instance_get(self):
        return getattr(self, json_column.key)[name]

    def instance_set(self, value):
        json_obj = getattr(self, json_column.key)
        if json_obj is None:
            setattr(self, json_column.key, {})
        getattr(self, json_column.key)[name] = value

    def cls_get(cls):
        expr = json_column[name]
        if type_:
            expr = expr.astext.cast(type_)
        return expr.label(prop_key)

    prop = hybrid_property(
        fget=instance_get,
        expr=cls_get,
        fset=instance_set
    )

    prop.__json_property = (json_column.key, name)
    prop.cls_get = cls_get
    prop.cls_get.prop_key = prop_key

    return prop


class ModelMetaType(DeclarativeMeta):
    def __new__(cls, name, bases, attrs):
        update_json_property_names(attrs)
        return DeclarativeMeta.__new__(cls, name, bases, attrs)



Base = declarative_base(metaclass=ModelMetaType)


class Author(Base):
    __tablename__ = "author"

    id = Column(Integer, primary_key=True)
    data = Column(JSONB, nullable=False)

    born = json_property(data, "yearBorn", Integer())
    died = json_property(data, "yearDied")

    def __repr__(self):
        return "Author(id={!r}, data={!r})".format(
            self.id, self.data
        )

def test_expression_has_name_of_json_property(session):
    session.add(Author(data={"yearBorn": 1881}))
    session.commit()
    q = select(Author.born)
    result = session.execute(q).one()
    assert result._asdict() == {"born": 1881}

原理说明 #

SQLAlchemy 1.4对查询结果的列名解析逻辑更严格，未指定别名的表达式会被自动命名为anon_*。通过在类级表达式中调用.label(prop_key)，并将属性名绑定到prop_key，我们强制将查询结果的列名设置为属性名称，从而让_asdict()返回预期的键值对。

内容的提问来源于stack exchange，提问作者Andrew