如何在Graphene DjangoFilterConnectionField查询集集成自定义输入参数？

我想知道如何以最简洁的方式将Graphene DjangoFilterConnectionField的自定义输入参数集成到默认查询集中？

我正在编写一个Graphene Query类，尝试通过自定义输入属性集成PgVector的CosineDistance函数，代码如下：

class Query(graphene.ObjectType):

    all_the_small_things = DjangoFilterConnectionField(
        VectorNode,
        embedding=graphene.List(graphene.Float),
    )

    def resolve_all_the_small_things(
        root,
        info,
        **kwargs,
    ):
        from pgvector.django import CosineDistance

        annotations = {}

        if kwargs.get("embedding"):
            annotations["embedding"] = CosineDistance(
                "embedding", kwargs.pop("embedding")
            )

        if annotations:
            # 如何最好地将这个逻辑集成到现有行为中？
            # VectorNode.objects.annotate(**annotations)
            # 现在要使用默认解析器提供的过滤功能。

我的需求是：当提供自定义输入时增强默认查询行为（比如过滤），未提供时保持默认行为不变。我是否需要实现自己的查询集和/或过滤器集？有没有办法通过函数参数调整默认行为？

注：我认为长期解决方案是为Graphene添加PG Vector列支持，或创建django-filters过滤器子类，但在考虑上游实现之前，我需要先理解这些组件的协作机制。

不需要完全重写查询集或过滤器集，你可以通过调用默认解析器并传入修改后的查询集实现需求，具体步骤如下：

1. 核心实现思路 #

先处理自定义参数（比如embedding）修改查询集，再将修改后的查询集交给DjangoFilterConnectionField的默认解析器，保留其原生的过滤、分页、游标查询等功能。

2. 修改后的完整代码 #

class Query(graphene.ObjectType):
    all_the_small_things = DjangoFilterConnectionField(
        VectorNode,
        embedding=graphene.List(graphene.Float),
    )

    def resolve_all_the_small_things(
        root,
        info,
        **kwargs,
    ):
        from pgvector.django import CosineDistance
        from graphene_django.filter.utils import get_filterset_class

        # 提取并移除自定义embedding参数
        embedding = kwargs.pop("embedding", None)
        queryset = VectorNode.objects.all()

        # 添加余弦距离注解（可选：按距离排序）
        if embedding:
            queryset = queryset.annotate(
                cosine_distance=CosineDistance("embedding", embedding)
            ).order_by("cosine_distance")

        # 初始化默认过滤器集，应用原有过滤条件
        filterset_class = get_filterset_class(VectorNode, DjangoFilterConnectionField.default_filterset_factory)
        filterset = filterset_class(
            {k: v for k, v in kwargs.items() if k not in ["first", "last", "before", "after", "order_by"]},
            queryset=queryset,
            request=info.context
        )

        # 验证过滤器合法性
        if not filterset.is_valid():
            raise Exception(filterset.errors)

        # 调用默认解析器处理分页、排序等逻辑
        return DjangoFilterConnectionField.connection_resolver(
            filterset.qs,
            root,
            info,
            **kwargs
        )

3. 优化：封装QuerySet方法简化代码 #

如果需要复用向量注解逻辑，可以给模型定义自定义QuerySet：

# 模型的自定义QuerySet
class VectorNodeQuerySet(models.QuerySet):
    def with_cosine_distance(self, embedding):
        return self.annotate(
            cosine_distance=CosineDistance("embedding", embedding)
        ).order_by("cosine_distance")

# 模型中绑定该QuerySet
class VectorNode(models.Model):
    # ... 你的字段定义
    objects = VectorNodeQuerySet.as_manager()

# 简化resolve方法
def resolve_all_the_small_things(root, info, **kwargs):
    from graphene_django.filter.utils import get_filterset_class

    embedding = kwargs.pop("embedding", None)
    queryset = VectorNode.objects.all()
    
    if embedding:
        queryset = queryset.with_cosine_distance(embedding)
    
    # 后续过滤器和默认解析器调用逻辑同上
    filterset_class = get_filterset_class(VectorNode, DjangoFilterConnectionField.default_filterset_factory)
    filterset = filterset_class(
        {k: v for k, v in kwargs.items() if k not in ["first", "last", "before", "after", "order_by"]},
        queryset=queryset,
        request=info.context
    )

    if not filterset.is_valid():
        raise Exception(filterset.errors)

    return DjangoFilterConnectionField.connection_resolver(
        filterset.qs,
        root,
        info,
        **kwargs
    )

关键说明 #

• 完全保留默认行为：通过调用connection_resolver，无需自己实现分页、游标查询等DjangoFilterConnectionField的核心功能。
• 逻辑叠加生效：先处理自定义向量注解，再应用原有过滤条件，确保两者同时生效。
• 无需自定义过滤器集：如果只是添加注解和排序，不需要扩展django-filters；后续若需基于余弦距离做范围过滤，再考虑创建过滤器子类即可。

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