时间戳

首页 > 解决方案 > 在 SQLAlchemy 中使用 Postgres 索引运算符

python - 在 SQLAlchemy 中使用 Postgres 索引运算符

问题描述

给定具有以下架构的表：

create table json_data (
    id integer PRIMARY KEY NOT NULL,
    default_object VARCHAR(10) NOT NULL,
    data jsonb NOT NULL
);

对于表中的每个实体，我想检索data['first']['name']字段的值，或者如果它是 null 值data[json_data.default_object]['name']，或者如果后者也是 null 然后返回一些默认值。在“纯”SQL 中，我可以编写以下代码来满足我的需要：

insert into
  json_data(
    id,
    default_object,
    data
  )
  values(
    0,
    'default',
    '{"first": {"name": "first_name_1"}, "default": {"name": "default_name_1"}}'
  ),
  (
    1,
    'default',
    '{"first": {}, "default": {"name": "default_name_2"}}'
  );

select
  id,
  coalesce(
    json_data.data -> 'first' ->> 'name',
    json_data.data -> json_data.default_object ->> 'name',
    'default_value'
  ) as value
from
  json_data;

我试图将上面的“模型”“翻译”成一个 SQLAlchemy 实体：

import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property


Base = declarative_base()


class JsonObject(Base):
    __tablename__ = 'json_data'

    id = sa.Column(sa.Integer, primary_key=True)
    default_object = sa.Column(sa.String(10), nullable=False)
    data = sa.Column(postgresql.JSONB, nullable=False)

    @hybrid_property
    def name(self) -> str:
        obj = self.data.get('first')
        default_obj = self.data.get(self.default_object)
        return (obj.get('name') if obj else default_obj.get('name')) or default_obj.get('name')

    @name.setter
    def name(self, value: str):
        obj = self.data.setdefault('first', dict())
        obj['name'] = value

    @name.expression
    def name(self):
        return sa.func.coalesce(
            self.data[('first', 'name')].astext,
            self.data[(self.default_object, 'name')].astext,
            'default_value',
        )

但似乎name混合属性的表达式不像我预期的那样工作。name如果我按属性查询实体，例如：

query = session.query(JsonObject).filter(JsonObject.name == 'name')

由querySQLAlchemy 扩展为如下内容：

SELECT json_data.id AS json_data_id, json_data.default_object AS json_data_default_object, json_data.data AS json_data_data 
FROM json_data 
WHERE coalesce((json_data.data #> %(data_1)s), (json_data.data #> %(data_2)s), %(coalesce_1)s) = %(coalesce_2)s

它使用路径运算符而不是索引运算符。我应该怎么做才能让 SQLAlchemy 创建一个表达式，就像我在问题开头写的那样？

标签： pythonpostgresqlsqlalchemy

解决方案

好的，我找到的解决方案非常简单。正如 SQLAlchemy 文档所说：

索引操作返回一个表达式对象，其类型默认为 JSON，因此可以在结果类型上调用进一步的面向 JSON 的指令。

因此，我们可以使用“链式”python 索引运算符。所以下面的代码在我看来是合法的：

class JsonObject(Base):
    # Almost the same stuff, except for the following:
    @name.expression
    def name(self):
        return sa.func.coalesce(
            self.data['first']['name'].astext,
            self.data[self.default_object]['name'].astext,
            'default_value',
        )

推荐阅读