You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

实现支持泛型的@add_filter装饰器以简化自定义过滤器集合的方法定义

实现支持泛型的@add_filter装饰器以简化自定义过滤器集合的方法定义

咱们先来梳理下你的核心需求:你希望用一个装饰器自动生成带exclude参数的过滤器方法,不用每次都写嵌套函数和重复调用add的冗余代码,同时还要保证类型提示准确,IDE能正常识别方法参数。之前的尝试要么有类型不兼容问题,要么丢失了参数的智能提示,下面咱们一步步解决这些问题。

第一步:明确类型变量与基类定义

首先我们需要定义清晰的泛型类型变量,确保装饰器能正确关联过滤器集合和目标对象的类型:

from typing import (
    Generic, TypeVar, Callable, Concatenate, ParamSpec, Self,
    partial, wraps
)
import inspect

# 目标对象的类型
T = TypeVar("T")
# 过滤器方法的额外参数(除了self和obj之外的参数)
P = ParamSpec("P")
# 过滤器集合的类型,绑定到Filters基类
FilterCollection = TypeVar("FilterCollection", bound="Filters")

然后是你的基础FnColl类,我们完善下_invert方法的实现:

class FnColl(Generic[T]):
    def __init__(self, fns: list[Callable[[T], Any]] = []) -> None:
        self.fns: list[Callable[[T], Any]] = fns

    def add(self, fn: Callable[[T], Any], invert: bool) -> Self:
        if invert:
            fn = self._invert(fn)
        self.fns.append(fn)
        return self
    
    @staticmethod
    def _invert(fn: Callable[[T], Any]) -> Callable[[T], Any]:
        def inverted(obj: T) -> Any:
            # 针对过滤器的bool返回值取反
            return not fn(obj)
        return inverted

第二步:实现带类型提示的@add_filter装饰器

我们把装饰器定义在Filters基类中,利用描述符模式处理实例绑定,同时保留原方法的参数签名,让IDE能正确识别参数:

class Filters(FnColl[T], Generic[T]):
    def __init__(self, fns: list[Callable[[T], bool]] = []):
        super().__init__(fns=fns)

    def add(self, fn: Callable[[T], bool], invert: bool) -> Self:
        return super().add(fn=fn, invert=invert)

    @staticmethod
    def add_filter(
        func: Callable[Concatenate[FilterCollection, T, P], bool]
    ) -> Callable[Concatenate[FilterCollection, bool, P], FilterCollection]:
        class FilterDescriptor(Generic[FilterCollection, T, P]):
            def __init__(self, func: Callable[Concatenate[FilterCollection, T, P], bool]):
                # 保留原函数的元数据(名称、文档字符串等)
                self.func = wraps(func)(func)

            def __get__(
                self, instance: FilterCollection, owner: type[FilterCollection]
            ) -> Callable[Concatenate[bool, P], FilterCollection]:
                def wrapper(exclude: bool = False, *args: P.args, **kwargs: P.kwargs) -> FilterCollection:
                    # 生成真正的过滤器函数:接收目标对象,调用原方法
                    def filter_func(obj: T) -> bool:
                        return self.func(instance, obj, *args, **kwargs)
                    # 添加到过滤器集合,支持链式调用
                    return instance.add(fn=filter_func, invert=exclude)
                
                # 手动修正包装函数的签名,让IDE能识别参数
                original_sig = inspect.signature(self.func)
                # 移除原方法的self和obj参数,添加exclude参数,保留其他参数
                new_params = [
                    inspect.Parameter(
                        "exclude",
                        inspect.Parameter.KEYWORD_ONLY,
                        default=False,
                        annotation=bool
                    ),
                    *list(original_sig.parameters.values())[2:]
                ]
                wrapper.__signature__ = original_sig.replace(parameters=new_params)
                return wrapper

        return FilterDescriptor(func)

第三步:编写具体的过滤器子类

现在你可以轻松定义针对具体对象的过滤器集合,不用再写重复的嵌套函数:

# 示例目标对象
class A:
    def __init__(self, x: int | None, y: str):
        self.x = x
        self.y = y

# 针对A的过滤器集合
class AFilters(Filters[A]):
    @Filters.add_filter
    def has_value_for_x(self, obj: A) -> bool:
        """检查A对象的x属性是否有值"""
        return obj.x is not None

    @Filters.add_filter
    def has_specific_value_for_y(self, obj: A, specific_value: str) -> bool:
        """检查A对象的y属性是否等于指定值"""
        return obj.y == specific_value

第四步:使用验证

现在你可以链式调用过滤器方法,IDE会正确提示exclude参数和原方法的参数:

# 创建过滤器实例并链式配置
filter_set = AFilters()\
    .has_value_for_x(exclude=True)\
    .has_specific_value_for_y(specific_value="abc")

# 测试过滤器
test_obj1 = A(x=None, y="abc")
test_obj2 = A(x=5, y="def")

def apply_filters(obj: A, filters: Filters[A]) -> bool:
    return all(fn(obj) for fn in filters.fns)

print(apply_filters(test_obj1, filter_set))  # 输出True:x为None(exclude=True取反后符合),y等于abc
print(apply_filters(test_obj2, filter_set))  # 输出False:x不为None(exclude=True取反后不符合)

关键细节解释

  1. 泛型与类型绑定:通过FilterCollection类型变量绑定到Filters基类,确保装饰器返回的是当前过滤器集合的实例,支持链式调用的类型提示。
  2. 参数签名保留:利用inspect模块手动修正包装函数的签名,让IDE能正确识别exclude参数和原方法的额外参数,解决了你之前提到的参数 introspection 问题。
  3. 描述符模式:通过__get__方法实现描述符,当实例访问装饰后的方法时,动态生成绑定到当前实例的包装函数,确保self参数正确传递。

备注:内容来源于stack exchange,提问作者DblDpl

火山引擎 最新活动