ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

django-filter 源码拆解:用 FilterSet 实现声明式查询参数过滤

django-filter 源码拆解:用 FilterSet 实现声明式查询参数过滤

简介:django-filters 作为 Django 生态中处理数据过滤的核心库,其源码包为 Python 后端开发者理解查询过滤机制提供了直接素材。它面向需要掌握 FilterSet 工作原理、自定义过滤器实现细节,以及如何在 Django REST Framework 中集成过滤后端的开发者。压缩包内含 61 个文件,16 个 py 源码文件涵盖 filterset、fields、filters、widgets 等核心模块,另有 15 个 pyc 编译文件与 14 个 po、13 个 mo 多语言翻译文件,整体仅 99KB,体积紧凑。通过研读各模块代码,读者可以厘清 CharFilter、NumberFilter、DateRangeFilter 等过滤器类型的内部实现,掌握 lookup_expr、order_by、distinct 等高级选项的作用机制,并理解自定义 Filter 时需要覆写 filter 方法的完整逻辑。该资源已有 289 人学习下载,适合正在学习 Django 源码、希望提升 API 接口数据筛选能力的开发者作为案头参考资料。

1. 拆 django-filters 源码包:Python 项目里最省心的声明式过滤方案

django-filters 这套源码包,拆开之后就是 Django 生态里处理「查询参数转 ORM 过滤条件」最成熟的一套框架。它解决的是列表接口最磨人的那部分:前端往 URL 里扔?status=paid&amount_min=100,后端要做的无非是把参数校验、lookup 拼接、queryset 过滤这些重复逻辑收敛起来。以前手动写if request.GET.get('amount_min'): queryset = queryset.filter(amount__gte=...),条件一多视图层全是样板代码。django-filters 的做法是声明式:定义一个 FilterSet 类,字段和查询表达式写清楚,剩下的交给框架。适合两类人:DRF 接口写得多、被重复过滤逻辑折磨的后端开发,以及想搞明白查询参数到底怎么一步步变成数据库 SQL 的初学者。

2. 架构拆解:FilterSet 类声明到 filters 字典的完整链路

2.1 包目录:django_filters 里每个模块各管什么

拆开源码包,django_filters/下文件不多,但分层非常清晰,我一般是按这个顺序去读的。

django_filters/ ├── __init__.py # 对外导出 Filter、FilterSet、各类内置 Filter ├── filterset.py # FilterSet 主体:元类、选项、过滤主循环 ├── filters.py # Filter 基类 + CharFilter/NumberFilter 等 ├── fields.py # Filter 与 Django Form Field 的桥接层 ├── widgets.py # 范围、多选等复合 widget ├── conf.py # 全局配置项 ├── utils.py # EMPTY_VALUES、label 生成等工具 ├── views.py # 非 DRF 场景的 FilterView 封装 └── rest_framework/ ├── backend.py # DjangoFilterBackend,DRF 的接入核心 ├── filterset.py # 带 request 的 rest_framework FilterSet └── views.py # FilterSetMixin,视图混入

filterset.py决定「怎么组织过滤声明」,filters.py决定「每个条件支持什么查询表达式」,rest_framework/backend.py决定「请求进来之后什么时候触发过滤」。读源码我建议从filterset.py入手,因为它里面那个元类解释了为什么 FilterSet 写起来像 Django ModelForm 一样舒服。

2.2 FilterSetMetaclass:类的声明怎么变成 base_filters 字典

FilterSet 的声明式体验来自元类。源码里FilterSetMetaclass.__new__的核心逻辑,我抽成可读版本:

# 提炼自 django_filters/filterset.py,省略了异常检查和属性拷贝 class FilterSetMetaclass(type): def __new__(cls, name, bases, attributes): filters = {} # 1. 先从父类继承已声明的过滤器 for base in bases: if hasattr(base, 'base_filters'): filters.update(base.base_filters) # 2. 把当前类里显式声明的过滤器并入 declared = attributes.pop('base_filters', {}) filters.update(declared) # 3. 生成新类并挂上元信息 new_class = super().__new__(cls, name, bases, attributes) new_class._meta = FilterSetOptions(attributes.get('Meta')) new_class.base_filters = filters return new_class

这一段的关键在于:父类的过滤器先收集,当前类的显式声明后覆盖,所以子类里重写同名 filter 是有效的。FilterSetOptions就是把Meta里的model、fields、exclude、filter_overrides、ordering这些配置解析成规范对象,给后续初始化用。

有一点容易误解:Meta.fields的展开并不在元类里做。元类只负责收集「显式写出来的 filter 属性」,而fields里声明{'total_amount': ['gte', 'lte']}这种展开,是在 FilterSet 实例化时通过setup_filters完成的。理解这一点,就不会在调试时奇怪「为什么元类跑完了 filters 还是空的」。

2.3 Filter 基类:field_name、lookup_expr、exclude、method、distinct

filters.py里的Filter是所有过滤器的基类,构造参数决定了每个条件的行为。

class Filter: def __init__(self, field_name=None, lookup_expr=None, *, exclude=False, method=None, label=None, distinct=False, **kwargs): self.field_name = field_name or self.field_name self.lookup_expr = lookup_expr or self.lookup_expr self.exclude = exclude self.method = method self.distinct = distinct self.label = label # kwargs 会传给表单字段构造器,比如 max_length、required

参数里最常用的是前两个:field_name支持点路径,比如customer__name表示关联表字段,这在联表过滤里几乎是刚需;lookup_expr默认exact,想模糊搜就传icontains。exclude=True时过滤动作从filter()变成exclude(),实现「不等于某值」这类反选需求。method是高级入口,传一个方法名,框架会调用filter_xxx(self, queryset, name, value),适合一个条件要拼多个 ORM 表达式的场景。distinct=True专门治关联过滤后行数翻倍的问题,后面避坑章节会细讲。

2.4 rest_framework/backend.py:DRF 请求在哪一步被拦下来

DRF 的接入点全部在rest_framework/backend.py,核心类DjangoFilterBackend继承 DRF 的BaseFilterBackend:

class DjangoFilterBackend(BaseFilterBackend): filterset_base = django_filters.FilterSet def get_filterset_class(self, view, queryset=None): if hasattr(view, 'filterset_class') and view.filterset_class: return view.filterset_class if hasattr(view, 'filterset_fields') and view.filterset_fields: return filterset_factory( queryset.model, fields=view.filterset_fields) return None def filter_queryset(self, request, queryset, view): filterset = self.get_filterset(request, queryset, view) if filterset is None: return queryset if not filterset.is_valid(): raise ValidationError(filterset.errors) return filterset.qs

这里的执行时机是 DRF 的ListView在get_queryset()之后、分页之前去调filter_backends,所以过滤结果天然携带到分页和序列化阶段。filter_queryset开头先生成 filterset_class,找不到就放行;找到之后先is_valid(),校验不过直接抛ValidationError,前端会收到字段级别的 400 错误。这也是它比手写过滤更规范的地方——参数错误不会悄悄变成空列表。

3. 快速跑通:从 pip 安装到第一个可过滤的 DRF 接口

3.1 安装与基础配置

安装就一条命令:

pip install django-filter

然后在settings.py里做两处配置。第一处是INSTALLED_APPS,我一般会加上django_filters,虽然纯 API 场景不强制,但它能让模板过滤标签和默认 widget 都可用,省得后面踩坑。第二处是 DRF 的全局过滤器配置。

INSTALLED_APPS = [ # ... 'django_filters', ] REST_FRAMEWORK = { 'DEFAULT_FILTER_BACKENDS': [ 'django_filters.rest_framework.DjangoFilterBackend', ], }

配置好之后,所有 DRF 视图都能直接用filterset_fields或filterset_class。如果只想让某些视图启用过滤,就别配全局的,改成在视图里单独写filter_backends。我维护的老项目里经常有多个自定义 backend,全局配置顺序一旦不对,过滤和排序会互相覆盖,这点避坑章会专门说。

3.2 定义第一个 FilterSet:字段声明与 lookup 映射

假设后台有一张订单表,结构简化如下:

# models.py class Customer(models.Model): name = models.CharField(max_length=64) class Order(models.Model): STATUS_CHOICES = [ ('pending', '待支付'), ('paid', '已支付'), ('shipped', '已发货'), ('done', '已完成'), ] customer = models.ForeignKey(Customer, on_delete=models.CASCADE, related_name='orders') status = models.CharField(max_length=16, choices=STATUS_CHOICES, db_index=True) total_amount = models.DecimalField(max_digits=10, decimal_places=2) created_at = models.DateTimeField(auto_now_add=True)

对应的 FilterSet 写在filters.py:

# filters.py import django_filters from .models import Order class OrderFilter(django_filters.FilterSet): # 关联字段过滤:customer__name 走 JOIN customer_name = django_filters.CharFilter( field_name='customer__name', lookup_expr='icontains') # 区间过滤:两个条件指向同一个模型字段 amount_min = django_filters.NumberFilter( field_name='total_amount', lookup_expr='gte') amount_max = django_filters.NumberFilter( field_name='total_amount', lookup_expr='lte') # 状态枚举,直接复用模型上的 choices status = django_filters.ChoiceFilter(choices=Order.STATUS_CHOICES) class Meta: model = Order fields = { 'created_at': ['exact', 'lt', 'gt'], 'status': ['exact'], }

这里演示了两种声明方式:显式声明适合需要自定义参数名的场景,比如amount_min和amount_max对应同一个total_amount字段;Meta.fields字典式声明会把'created_at': ['exact', 'lt', 'gt']展开成created_at__exact、created_at__lt、created_at__gt三个过滤器,参数名直接带双下划线,URL 长这样:/api/orders/?created_at__gte=2024-01-01。

3.3 在 DRF 视图中启用并验证效果

视图里挂上filterset_class就行:

# views.py from django_filters.rest_framework import DjangoFilterBackend from rest_framework import generics from .models import Order from .serializers import OrderSerializer from .filters import OrderFilter class OrderListView(generics.ListAPIView): queryset = Order.objects.select_related('customer').all() serializer_class = OrderSerializer filter_backends = [DjangoFilterBackend] filterset_class = OrderFilter

启动服务后直接请求:

curl 'http://127.0.0.1:8000/api/orders/?status=paid&amount_min=100&customer_name=张'

status=paid精确匹配订单状态,amount_min=100转成total_amount__gte=100,customer_name=张变成customer__name__icontains='张',三个条件 AND 在一起。想确认 SQL 有没有按预期拼出来,在 FilterSet 实例上打印即可:

filterset = OrderFilter(request.GET, Order.objects.all()) print(str(filterset.qs.query))

马上能看到 WHERE 子句里的status = paid AND total_amount >= 100 AND customer__name LIKE %张%,这对排查联表条件非常有用。

3.4 三个高频配置:ordering、exclude、filter_overrides

排序需求在列表接口里太常见了,OrderingFilter 是单独的过滤器:

class OrderFilter(django_filters.FilterSet): ordering = django_filters.OrderingFilter( fields=( ('created_at', 'created_at'), ('amount', 'total_amount'), ) ) class Meta: model = Order fields = ['status', 'total_amount'] exclude = ['total_amount']

ordering参数里的元组是「URL 参数名 → 模型字段名」的映射,比如('amount', 'total_amount')表示 URL 写?ordering=amount,实际排序字段是total_amount。这样设计的好处是前端参数名跟模型字段解耦,后面改库字段名不影响接口。默认排序按ordering传入的-前缀反转,?ordering=-amount就是按金额倒序。

exclude用于排除不想暴露的字段,在fields之外做减法。filter_overrides是全局覆盖某类模型字段的默认过滤器行为,比如 DecimalField 默认是精确匹配,业务上列表接口几乎都是查区间,可以统一改成这样:

import django.db.models as models filter_overrides = { models.DecimalField: { 'filter_class': django_filters.NumberFilter, 'extra': lambda f: {'lookup_expr': 'gte'}, }, }

这里extra是个可调用对象,接收模型字段实例并返回额外参数,灵活度很高。不过这种全局覆盖会影响所有 ModelForm 生成的 FilterSet,适合团队约定,不适合个人项目随便全局替换。

4. 源码级追踪:一次过滤请求从 query 参数到 SQL 的完整链路

4.1 FilterSet.init与 qs 的懒加载

要理解过滤是怎么发生的,先看BaseFilterSet.__init__:

class BaseFilterSet: def __init__(self, data=None, queryset=None, *, request=None, prefix=None): if queryset is None: queryset = self._meta.model._default_manager.all() self.is_bound = data is not None self.data = data or {} self.queryset = queryset self.request = request self.filters = copy.deepcopy(self.base_filters) self.form = self.get_form() @property def qs(self): if not hasattr(self, '_qs'): qs = self.queryset.all() if self.is_bound: qs = self.filter_queryset(qs) self._qs = qs return self._qs

is_bound是理解整套逻辑的门槛:只有绑定了请求数据才算「正在过滤」,否则qs直接返回全量。self.filters是base_filters的深拷贝,避免实例之间互相污染,这也是官方推荐不在类属性上直接改filters的原因。qs用了懒加载加缓存,第一次访问时执行过滤并暂存在_qs上,后续再访问不会重复跑 SQL。

4.2 filter_queryset 主循环:每个 filter 依次触发

qs触发后的核心是filter_queryset:

def filter_queryset(self, queryset): for name, filter_ in self.filters.items(): value = self.form.cleaned_data.get(name) if value is not None: queryset = filter_.filter(queryset, value) return queryset

这段循环把「校验」和「过滤」分开:self.form是按 filters 动态生成的 Django Form,每个 Filter 会生成一个对应类型的表单字段,cleaned_data是校验通过后的干净值;校验不通过的值根本到不了这里,这也是为什么前面 backend 里要先is_valid()。循环按声明顺序执行,前面的条件先作用到 queryset,后面的条件在结果集上继续收窄。

有个细节值得注意:判断条件用的是is not None,不是if value。因为False、0都是合法过滤值,用 truthy 判断会把「传了 false 但想过滤出假值」的场景吞掉。这个坑后面会专门展开。

4.3 Filter.filter:值到 lookup 表达式的最后一跳

经过校验的值进入Filter.filter:

# 来自 django_filters/filters.py,保留核心逻辑 EMPTY_VALUES = ([], (), {}, '', b'', None) def filter(self, qs, value): if value in EMPTY_VALUES: return qs if self.distinct: qs = qs.distinct() lookup = '%s__%s' % (self.field_name, self.lookup_expr) qs = self.get_method(qs)(**{lookup: value}) return qs def get_method(self, qs): return qs.exclude if self.exclude else qs.filter

最核心的一行是lookup = '%s__%s' % (self.field_name, self.lookup_expr),它把customer__name和icontains拼成 Django ORM 能吃的查询表达式,然后qs.filter(**{lookup: value})展开执行。EMPTY_VALUES是空值白名单,空字符串、空列表、None 都直接放行,不参与过滤;但注意False和0不在里面,所以能正常过滤假值。exclude=True时get_method返回qs.exclude,实现反选。

4.4 一次完整请求的链路串讲

把前面几节串起来,一次带过滤的 DRF 请求完整走以下六个阶段:

阶段执行位置发生了什么
1. 参数解析DRF request.query_paramsURL 查询串变成 QueryDict
2. 后端触发DjangoFilterBackend.filter_queryset根据 filterset_class 生成 FilterSet 实例
3. 表单校验FilterSet.get_form() / is_valid()每个 Filter 生成表单字段,清洗并校验
4. 过滤主循环FilterSet.filter_queryset按声明顺序逐个调用 Filter.filter
5. 查询拼接Filter.filterlookup 拼接后执行 qs.filter/exclude
6. 结果返回DRF 分页与序列化filterset.qs 作为最终 queryset

中间任何一步出问题,定位路径也是反着来:先看后端有没有生成 FilterSet,再看is_valid()返回的错误信息,最后打印filterset.qs.query确认 SQL。这套链路只要你亲手追过一次,后面写任何自定义过滤器都不会再觉得是黑匣子。

5. 避坑手册:真实项目里最常见的 5 个过滤翻车点

5.1 现象一:按关联字段过滤后行数翻倍

列表接口加了items__product_name__icontains这类关联过滤后,返回结果里同一订单出现多次,分页 total 也对不上。

原因:一对多关联 JOIN 时,一个订单匹配多条子表记录,queryset 里会出现重复行;django-filters 默认不自动去重。

解决:声明过滤器时直接开distinct=True:

item_name = django_filters.CharFilter( field_name='items__product_name', lookup_expr='icontains', distinct=True, )

如果多个过滤器都要关联同一张表,或者自定义 method 里拼了多个 JOIN,我会在 FilterSet 里统一处理:

class OrderFilter(django_filters.FilterSet): def filter_queryset(self, queryset): return super().filter_queryset(queryset).distinct()

注意去重最好在过滤阶段做,等分页拿到结果集再 distinct 就晚了。

5.2 现象二:BooleanFilter 传 false 过滤不出来

前端给?is_active=false,接口返回的还是全量数据,日志里也没报错。

原因:很多人会在自定义方法或重写filter_queryset时写if value:这种 truthy 判断,False和0被当成假值跳过了。前面源码里也看到了,框架本身的判断是value is not None加EMPTY_VALUES白名单,False不在白名单里,应该正常过滤。

解决:自定义过滤方法里统一用is not None判断,并注意空字符串要放行:

def filter_active(self, queryset, name, value): if value in ([], (), {}, '', b'', None): return queryset return queryset.filter(is_active=value)

血泪经验:这个坑排查起来最费时间,因为不报错只差结果,最后是把self.form.cleaned_data打出来才发现值一直是False。

5.3 现象三:ChoiceFilter 传了枚举外的值直接 400

前端 Select 组件数据跟后端 choices 没对齐,传了个历史脏数据的状态值,接口直接返回 400,错误信息是「Select a valid choice」。

原因:ChoiceFilter生成的是 DjangoChoiceField,校验规则严格按 choices 白名单来。模型里有脏数据、或者前端枚举配置过期,都会触发校验失败。

解决:如果这个条件只是透传给数据库过滤,不参与表单校验,直接绕开 ChoiceFilter:

status = django_filters.CharFilter( field_name='status', lookup_expr='exact')

想保留校验又不想 400,可以在视图里捕获ValidationError统一返回空列表,或者在前端枚举接口里做数据对齐。我一般倾向用 CharFilter 加自定义 method,白名单逻辑自己控制。

5.4 现象四:filter_backends 顺序不对导致过滤互相覆盖

同时配了SearchFilter和DjangoFilterBackend,结果排序和过滤的结果总是不符合预期,有时候搜索条件直接丢了。

原因:DRF 的filter_backends按列表顺序依次调用,每个 backend 的filter_queryset拿到的是上一个 backend 处理后的 queryset。DjangoFilterBackend内部的get_filterset生成 FilterSet 时用的是当前 view 的 queryset 属性,如果前面 backend 已经换掉了 queryset 而 FilterSet 还拿原始 queryset 去过滤,就会出现条件丢失。

解决:把全局配置里的顺序固定为「先基础过滤,再做搜索」,并统一在 view 里通过get_queryset()维护基础范围:

REST_FRAMEWORK = { 'DEFAULT_FILTER_BACKENDS': [ 'django_filters.rest_framework.DjangoFilterBackend', 'rest_framework.filters.SearchFilter', 'rest_framework.filters.OrderingFilter', ], }

5.5 现象五:大表模糊查询把线上接口拖垮

订单量过百万后,customer_name__icontains查询延迟从 50ms 涨到两秒多,数据库 CPU 被打满。

原因:icontains走的是LIKE '%keyword%',前导通配符让普通 B-tree 索引完全失效,全表扫描一次就是灾难。

解决:先看执行计划确认索引情况,再调整查询表达式。常见的做法是改为istartswith匹配前缀,能走索引;真正需要全文搜索的场景,换数据库层方案而不是在过滤器里硬扛。调试时用这条命令看 SQL:

print(str(filterset.qs.query))

我一般会顺手EXPLAIN一下,确认type不是ALL。玄学调参不如把索引和查询类型搞清楚。

6. 二次开发:给团队封装一个跨字段搜索 Filter 与通用 FilterSet

6.1 自定义一个跨字段搜索 Filter

业务上最常遇到的需求是「一个关键字搜多个字段」。基于CharFilter扩展一个:

import django_filters from django.db.models import Q class MultiFieldSearchFilter(django_filters.CharFilter): def __init__(self, fields=None, *args, **kwargs): self.search_fields = list(fields or []) super().__init__(*args, **kwargs) def filter(self, queryset, value): if not value: return queryset conditions = Q() for field in self.search_fields: conditions |= Q(**{f'{field}__icontains': value}) return queryset.filter(conditions)

使用方式:

class OrderFilter(django_filters.FilterSet): keyword = MultiFieldSearchFilter( fields=['customer__name', 'items__product_name'])

这样一个keyword参数就能跨订单客户和商品名两个方向搜索,视图层不需要任何改动。

6.2 抽一个带租户隔离的 FilterSet 基类

多租户系统里每个请求都要限定数据范围。与其在每个视图里拼条件,不如在 FilterSet 基类统一处理:

from django_filters import FilterSet from django_filters.rest_framework import DjangoFilterBackend class TenantFilterBackend(DjangoFilterBackend): def filter_queryset(self, request, queryset, view): queryset = queryset.filter(tenant_id=request.tenant_id) return super().filter_queryset(request, queryset, view)

公司基建层提供这个 backend,业务代码只写过滤声明,不用关心租户隔离。注意super().filter_queryset生成的 FilterSet 必须基于同一个 queryset 来源,顺序不对会丢掉租户条件——避坑章第五节的排查思路在这里同样适用。

6.3 一套可复用的验证习惯

改完过滤器之后,我每次都会先写两条测试再手动验证:

from rest_framework.test import APITestCase from django.urls import reverse class OrderFilterTests(APITestCase): def test_amount_range(self): url = reverse('order-list') + '?amount_min=100&amount_max=200' resp = self.client.get(url) self.assertEqual(resp.status_code, 200) for order in resp.data['results']: self.assertTrue(100 <= order['total_amount'] <= 200) def test_empty_param_returns_all(self): resp = self.client.get(reverse('order-list')) self.assertEqual(resp.status_code, 200)

从那以后,我每次给新列表接口加过滤条件,都强制自己先写一条「参数没传时返回全量」的用例,再写实际过滤断言,SQL 查询也顺手打出来看一遍。django-filter 这套源码包里值得抄的不只是 FilterSet,它处理空值、校验、去重的那套边界逻辑,才是真正值钱的部分。希望帮到你。

本文还有配套的精品资源,点击获取

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进