在 django 查询集中横向加入（为了使用 jsonb_to_recordset postgresql 函数）

Question

我有一个模型“myModel”将一些数据保存在一个名为“json”的（postgresql）jsonField中，json数据的典型结构是：{key:[{"a":1, "b":2}, {" a":3, "b":4}]}。

我想根据“a”或“b”的值过滤 myModel 查询集。我可能还想聚合“a”或“b”

因此，“取消嵌套” (json -> key) 数组将非常受欢迎，但我无法弄清楚如何使用 django api 执行此操作。

我尝试通过以下 SQL 查询在 postgresql 中直接执行“取消嵌套”。

SELECT * 
FROM "myModel"
join lateral jsonb_to_recordset("myModel"."json" -> 'key') as r("a" int, "b" int) on true
LIMIT 5

我们甚至可以使用横向连接的快捷表示法使其更加紧凑

SELECT * 
FROM "myModel", jsonb_to_recordset("myModel"."json" -> 'key') as r("a" int, "b" int)
LIMIT 5

但我不知道如何使用 django API 做一些等效的事情。我已经尝试了一些关于 annotate 和 RawSQL 的事情，但他们似乎没有对“FROM”子句起作用。这是我应该实际添加 'jsonb_to_recordset' 语句的地方。我可能可以使用原始函数来放置我的原始 SQL，但这意味着我无法使用 django API 在连接的 quesryset 上“过滤”或“聚合”......我必须在 rawSQL 中做所有事情这对我必须做的事情来说不是很方便。

另一种方法是使用允许在 sql FROM 子句中添加附加表的查询集“额外”功能。不幸的是，如果我这样做：

qs = myModel.objects.all()
qs = qs.extra(tables = ["""jsonb_to_recordset("myApp_myModel"."json" -> 'key') as r("a" int, "b" int)"""])
qs = qs.values()
print(qs.query)

我得到查询 django 将执行：

SELECT * 
FROM "myModel", "jsonb_to_recordset("myModel"."json" -> 'key') as r("a" int, "b" int)"

这非常接近我需要的......除了django在我提供的额外“表”名称周围添加了额外的引号......所以该功能不再起作用。

知道如何处理这个问题吗？

提前致谢，洛伊克

Answer 1

尽管自从您询问这个问题以来已经晚了近 6 个月，但我仍在尝试为您的问题添加我的贡献。

最近，我一直在研究使用 Django ORM 来使用的 POSTGRESQL Jsonb 函数，目的是为 Django 构建强大的报告引擎库，我搜索并找到了您的问题陈述。看来我完全陷入了这个100%相同的问题。 How could I use JSON array's to apply aggregation/annotation functions for easy reporting to front-end and show graphs and tables?

经过三天的摸索、摸索、摸索，终于找到了解决的办法。

这可能不是一个理想的方式，但我尽力保持它尽可能理想。还试图避免任何可能的 SQL 注入。我们不要因为“谈论我的事情”而让我们的时代变得无聊。

现在就进入下面的实现：

from django.db.models.constants import LOOKUP_SEP
from django.db.models.sql.datastructures import BaseTable


class JsonbFunction:
    JSONB_TO_RECORDSET = ("jsonb_to_recordset", '#>')


class JsonbFunctionTable(BaseTable):
    jsonb_join_type = ("JOIN LATERAL", "ON TRUE")
    function_name = None
    function_alias = None

    def __init__(self, table_name, function_name, field_name, columns):
        field_name_seq = field_name.split(LOOKUP_SEP)
        self.model_field = field_name_seq[0]
        self.json_path = field_name_seq[1:]
        alias = self.model_field + 's'
        super(JsonbFunctionTable, self).__init__(table_name=table_name, alias=alias)
        self.function_name = function_name
        column_definitions = list()
        for _c in columns:
            column_definitions.append('{c_name} {c_type}'.format(c_name=_c[0], c_type=_c[1]))
        self.function_alias = '{alias}({column_definitions})'.format(
            alias=alias, column_definitions=','.join(column_definitions))

    def as_sql(self, compiler, connection):
        return "{join} {function}({table}.{field} {sign} '{json_path}') {f_alias} {condition}".format(
            join=self.jsonb_join_type[0],
            condition=self.jsonb_join_type[1],
            f_alias=self.function_alias,
            function=self.function_name[0],
            sign=self.function_name[1],
            table=compiler.quote_name_unless_alias(self.table_name),
            field=compiler.quote_name_unless_alias(self.model_field),
            json_path='{' + ",".join(self.json_path) + '}'
        ), []

    def relabeled_clone(self, change_map):
        return self.__class__(self.table_name, change_map.get(self.table_alias, self.table_alias))

    def equals(self, other, with_filtered_relation):
        return (
                isinstance(self, other.__class__) and
                self.table_name == other.table_name and
                self.table_alias == other.table_alias and
                self.function_name == other.function_name and
                self.join_type == other.join_type
        )

解释：

对于jsonb_to_recordset函数，它在内部生成一个表，为了得到我们想要的，我们必须将该函数的返回表与 Django 模型的每个相应行连接起来。为此，唯一的选择是允许LATERAL JOINwith jsonb_to_recordset returned table。

JsonbFunctionTable扩展 BaseTable 类，该类将允许您使用 django 查询集推送自定义 TABLE JOIN。

现在让我们跳到下一个代码段：

from django.contrib.postgres.fields import JSONField
from django.db.models import F, CharField, Expression
from django.db.models.constants import LOOKUP_SEP
from django.utils.deconstruct import deconstructible


class ForceColumn(Expression):
    """
    Represents the SQL of a column name without the table name.

    This variant of Col doesn't include the table name (or an alias) to
    avoid a syntax error in check constraints.
    """
    contains_column_references = True

    def __init__(self, target, output_field=None):
        if output_field is None:
            output_field = target
        super().__init__(output_field=output_field)
        self.target = target

    def __repr__(self):
        return '{}({})'.format(self.__class__.__name__, self.target)

    def as_sql(self, compiler, connection):
        return self.target.db_column, []

    def get_group_by_cols(self):
        return [self]

    def get_db_converters(self, connection):
        if self.target == self.output_field:
            return self.output_field.get_db_converters(connection)
        return (
                self.output_field.get_db_converters(connection) +
                self.target.get_db_converters(connection)
        )


@deconstructible
class ForceF(F):
    model = None
    json_field = None

    def __init__(self, model, name, json_field):
        super(ForceF, self).__init__(name)
        self.model = model
        self.json_field = json_field

    def resolve_expression(self, query=None, allow_joins=True, reuse=None,
                           summarize=False, for_save=False, simple_col=False):
        _field_ref = self.json_field + 's.' + self.name.replace(LOOKUP_SEP, '.')
        path, final_field, targets, rest = query.names_to_path(
            [self.json_field], query.get_meta(), query.get_initial_alias())
        if not isinstance(final_field, JSONField):
            raise Exception('`ForeF` only available for JSON Fields')
        _dummy_field = CharField(db_column=_field_ref, name=_field_ref)
        _dummy_field.model = self.model
        return ForceColumn(_dummy_field, _dummy_field)

解释

这里我定义了一个 ForceF 表达式函数来扩展 django 的F. 这样做的原因是，在查询中，您必须使用动态生成的 JSONField 的“值路径”来选择/order_by/group_by。但正如预期的那样，Django 不会允许你这样做，因为 Django 会尝试使用常规可用的 Django 字段来验证你的“动态生成的 JSON 列名称”。这会给你带来另一次堵塞的打击。并ForeceF在这里解决这个问题。它还通过强制生成表的命名约定规则来处理有意可能的 SQL 注入。

以下是您最终将如何使用 Django 查询集：

results = myModel.objects.filter()
# Adding extra JOIN for JSONb
join_config = JsonbFunctionTable(
    table_name=myModel._meta.db_table,
    function_name=JsonbFunction.JSONB_TO_RECORDSET,
    field_name='json__key', # Your JSON Field and path to that array
    columns=[('a', 'int'), ('b', 'int')]
)
results.query.join(join=join_config)
# Group by as your need
results = results.values('group_by_somethings')
# Now finally force annotate with your dynamically generated JSON Columns
results = results.annotate(
    jsonb_annotated=ForceF(model=myModel, name='a', json_field='json')
)
results = results.values("value_1", "value_2", "jsonb_annotated")
# Check generated query
print(results.query)
# Check generated result
print(results.query)

脚注：

到目前为止，这是我根据自己的需要构建的东西。稍后可能会产生不良后果，但目前它正在按预期工作，满足我的需要。所以，也许它不能满足您的确切要求，但我很确定这个实现可以扩展来解决您提到的问题。

无论谁读到这个疯狂的或可能是坏主意的人，一定不要忘记向我们表达您的担忧。