Django注解返回纯PK而非HashID的优化方案咨询

背景 #

应用中用户分为3种角色，每个角色对应专属的项目关联字段：

• PROGRAM_MANAGER → managed_programs
• SUPERVISOR → supervised_programs
• CASE_MANAGER → case_managed_programs

最初采用annotate方式查询所有用户及关联项目，仅产生5次数据库查询，但Program的HashID主键返回的是纯整数值；改用SerializerMethodField能返回正确HashID，但未复用prefetch_related预取的数据，导致查询量激增。

现有临时方案 #

已在序列化器中手动转换HashID，代码如下：

programs = serializers.SerializerMethodField()

@staticmethod
def get_programs(obj):
    return [
        {"id": str(Hashid(value=program['id'], salt=settings.SECRET_KEY, min_length=13)), "name": program['name']} for program in obj.programs
    ]

优化方案 #

方案1：保留注解方式，直接返回HashID #

要让annotate直接生成HashID字符串，可通过自定义数据库函数实现（以PostgreSQL为例）：

1. 创建数据库自定义函数
   将HashID的Python生成逻辑转为PL/pgSQL函数，执行以下SQL：

CREATE OR REPLACE FUNCTION hashid_encode(id integer, salt text, min_length integer)
RETURNS text AS $$
DECLARE
    alphabet text := 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890';
    base integer := length(alphabet);
    hashid text := '';
    remainder integer;
BEGIN
    IF id = 0 THEN
        RETURN substr(alphabet, 1, 1);
    END IF;
    WHILE id > 0 LOOP
        remainder := id % base;
        id := floor(id / base);
        hashid := substr(alphabet, remainder + 1, 1) || hashid;
    END LOOP;
    -- 应用salt混淆逻辑（需与Python端Hashid逻辑完全一致）
    hashid := hashid || salt;
    -- 补全最小长度
    IF length(hashid) < min_length THEN
        hashid := hashid || substr(alphabet, 1, min_length - length(hashid));
    END IF;
    RETURN hashid;
END;
$$ LANGUAGE plpgsql;

2. 在Django中定义对应的Func类

from django.db.models import Func, CharField

class HashidEncode(Func):
    function = 'HASHID_ENCODE'
    output_field = CharField()

3. 修改annotate逻辑
   在Subquery中使用自定义函数生成HashID字符串，替换原有的F('id')：

def get_queryset(self):
    queryset = (
        User.objects
        .select_related('profile')
        .prefetch_related('managed_programs', 'supervised_programs', 'case_managed_programs')
        .annotate(
            programs=Case(
                When(
                    role=RoleChoices.PROGRAM_MANAGER,
                    then=ArraySubquery(
                        Program.objects.filter(program_managers=OuterRef('pk'), is_active=True)
                        .annotate(
                            data=Func(
                                Value('{"id": "'),
                                HashidEncode(F('id'), Value(settings.SECRET_KEY), Value(13)),
                                Value('", "name": "'),
                                F('name'),
                                Value('"}'),
                                function='concat',
                                output_field=CharField(),
                            )
                        )
                        .values('data')
                    ),
                ),
                When(
                    role=RoleChoices.SUPERVISOR,
                    then=ArraySubquery(
                        Program.objects.filter(supervisors=OuterRef('pk'), is_active=True)
                        .annotate(
                            data=Func(
                                Value('{"id": "'),
                                HashidEncode(F('id'), Value(settings.SECRET_KEY), Value(13)),
                                Value('", "name": "'),
                                F('name'),
                                Value('"}'),
                                function='concat',
                                output_field=CharField(),
                            )
                        )
                        .values('data')
                    ),
                ),
                When(
                    role=RoleChoices.CASE_MANAGER,
                    then=ArraySubquery(
                        Program.objects.filter(case_managers=OuterRef('pk'), is_active=True)
                        .annotate(
                            data=Func(
                                Value('{"id": "'),
                                HashidEncode(F('id'), Value(settings.SECRET_KEY), Value(13)),
                                Value('", "name": "'),
                                F('name'),
                                Value('"}'),
                                function='concat',
                                output_field=CharField(),
                            )
                        )
                        .values('data')
                    ),
                ),
                default=Value(list()),
                output_field=ArrayField(base_field=JSONField()),
            )
        )
        .order_by('id')
    )
    return queryset

这样注解返回的programs字段直接包含正确的HashID字符串，无需在序列化器中二次转换。

方案2：让SerializerMethodField复用预取数据 #

原代码使用values_list()会触发新的数据库查询，改为遍历预取的对象实例即可复用缓存数据：

@staticmethod
def get_programs(obj):
    role_attr = {
        RoleChoices.PROGRAM_MANAGER: 'managed_programs',
        RoleChoices.SUPERVISOR: 'supervised_programs',
        RoleChoices.CASE_MANAGER: 'case_managed_programs',
    }
    try:
        programs_qs = getattr(obj, role_attr[obj.role], None)
        if not programs_qs:
            return []
        # 直接遍历预取的对象，而非调用values_list触发新查询
        return [
            {'id': str(program.id), 'name': program.name} 
            for program in programs_qs.all()
        ]
    except (AttributeError, KeyError):
        return []

若要在数据库层面过滤is_active=True的项目，可使用Prefetch对象指定预取条件：

from django.db.models import Prefetch

def get_queryset(self):
    # 为每个关联字段指定预取时的过滤条件
    managed_prefetch = Prefetch(
        'managed_programs',
        queryset=Program.objects.filter(is_active=True)
    )
    supervised_prefetch = Prefetch(
        'supervised_programs',
        queryset=Program.objects.filter(is_active=True)
    )
    case_managed_prefetch = Prefetch(
        'case_managed_programs',
        queryset=Program.objects.filter(is_active=True)
    )
    
    queryset = (
        User.objects
        .select_related('profile')
        .prefetch_related(managed_prefetch, supervised_prefetch, case_managed_prefetch)
        .order_by('id')
    )
    return queryset

这样序列化器中无需再过滤is_active，进一步提升效率。

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