如何让Django最近整数匹配函数适配JSONField中的bigint？

我有一个超大规模数据库，每个条目包含一个JSON字段，该字段存储约30个子对象，结构如下：

{
    "field1": {"someOther": "stuffInside", "target_bigint": 12345},
    "field2": {"someOther": "stuffInside", "target_bigint": 67890},
    [...]
}

所有子对象都包含一个bigint类型字段。我需要根据给定整数匹配最接近的条目，为此实现了如下函数，它对普通int字段工作正常，但无法适配JSONField：

def get_closest_match(
        field: str,
        match: int,
        model: django.db.models.Model,
        return_all_match: bool = False,
        sub_filter: dict or None = None,
):

    log.debug(f"Finding nearest in {model.__name__} with {field} {match}")

    result = model.objects.annotate(abs_diff=Func(F(field) - match, function='ABS'))

    if sub_filter:
        result = result.filter(**sub_filter)

    result = result.order_by('abs_diff')

    if not return_all_match:
        return result.first()

    return result

我最初考虑新建单独表存储这些bigint，但每个条目有近30个子对象且数据库有数百万条目，会导致性能问题和数据冗余，因此想找到更优方案。

解决方案 #

可以通过修改函数中的字段提取逻辑，利用Django的数据库函数从JSONField中提取bigint字段并计算差值，无需额外建表。以下是具体修改方案：

1. 导入必要的数据库函数 #

首先需要导入Django用于JSON提取、类型转换的内置函数：

from django.db.models import Func, F, Cast, BigIntegerField
from django.db.models.functions import JSONExtract
from django.db import connection
import logging

2. 完整修改后的函数 #

整合适配逻辑后，完整函数如下：

log = logging.getLogger(__name__)

def get_closest_match(
        field: str,
        match: int,
        model: django.db.models.Model,
        json_field_name: str = "data",  # 指定模型中JSONField的字段名，默认值为data
        return_all_match: bool = False,
        sub_filter: dict or None = None,
):

    log.debug(f"Finding nearest in {model.__name__} with JSON path {field} matching {match}")

    # 根据数据库引擎选择对应的JSON提取方式
    with connection.cursor() as cursor:
        db_engine = cursor.db.vendor

    if db_engine == 'postgresql':
        # PostgreSQL下使用jsonb_extract_path_text提取JSON值并转bigint
        result = model.objects.annotate(
            target_value=Cast(
                Func(F(json_field_name), field.replace('.', ','), function='jsonb_extract_path_text'),
                BigIntegerField()
            ),
            abs_diff=Func(F('target_value') - match, function='ABS')
        )
    elif db_engine == 'mysql':
        # MySQL下使用JSON_EXTRACT提取JSON值并转bigint
        result = model.objects.annotate(
            target_value=Cast(
                Func(F(json_field_name), f"'$.{field}'", function='JSON_EXTRACT'),
                BigIntegerField()
            ),
            abs_diff=Func(F('target_value') - match, function='ABS')
        )
    else:
        # 通用方案（Django 3.1+支持），使用JSONExtract跨数据库提取
        result = model.objects.annotate(
            target_value=Cast(
                JSONExtract(json_field_name, field),
                BigIntegerField()
            ),
            abs_diff=Func(F('target_value') - match, function='ABS')
        )

    if sub_filter:
        result = result.filter(**sub_filter)

    result = result.order_by('abs_diff')

    if not return_all_match:
        return result.first()

    return result

关键说明 #

• 新增json_field_name参数，用于指定模型中存储子对象的JSONField字段名，默认值为data，可根据实际情况调整
• 通过Cast函数将JSON中提取的值强制转换为BigIntegerField，确保数值计算的准确性
• 针对PostgreSQL和MySQL做了适配，避免跨数据库的语法差异问题
• 完全保留原函数的筛选、排序、返回逻辑，无需改动原有调用方式

这样修改后，函数可以直接处理JSONField中的bigint字段，无需额外建表，彻底避免数据冗余和性能损耗。

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