尽管你问这个问题已经晚了将近 6 个月，试图将我的贡献添加到你的问题中。
虽然最近我一直在研究 POSTGRESQL Jsonb 函数以使用 Django ORM 来为 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?
经过 3 天的反复试验，我终于在书房的墙壁上摸索，找到了解决问题的方法。
这可能不是一个理想的方式，但我尽量保持理想。还试图避免任何可能的 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 JOIN 和 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)
说明
这里我定义了一个扩展django的F的ForceF表达式函数。这样做的原因是，在 Query 中，您必须使用动态生成的 JSONField 的“您的价值路径”来选择/order_by/group_by。但正如预期的那样，Django 不允许您这样做，因为 Django 将尝试使用您的“动态生成的 JSON 列名称”和常规可用的 Django 字段进行验证。那会给你又一次堵塞的打击。 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)
脚注：
到目前为止，这是我为满足我的需要而构建的。以后可能会产生不良后果，但现在它正在按预期工作，以满足我的需要。所以，也许它不能满足你的确切要求，但我很确定这个实现可以扩展来解决你提到的问题。
而谁读到这个疯狂或可能是坏主意，一定不要忘记向我们表达您的担忧。