You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python:导入时用元类实现Filter Factory的方案是否Pythonic?求优化

过滤器自动注册工厂的实现与优化疑问

我希望在模块中新增过滤器时能简单实现,让它在导入时自动被库识别。比如调用FilterFactory.available_filters能获取所有过滤器的映射:

>>> FilterFactory.available_filters 
{
    'upper': __main__.FilterUpper,
    'lower': __main__.FilterLower,
    'trim': __main__.FilterTrim
}

最初的实现方案:classmethod + LRU Cache

我一开始用类方法结合LRU缓存的方式实现:

class FilterFactory:
    @classmethod
    @lru_cache()
    def available_filters(cls):
        fmap = {}
        for _, member in inspect.getmembers(sys.modules[__name__]):
            if not inspect.isclass(member) or not hasattr(member, 'name'):
                continue
            if member.name() == 'base':
                continue
            fmap[member.name()] = member
        return fmap

改进方案:元类实现自动注册

后来我觉得用元类在模块加载时构建工厂更合适,于是实现了以下代码:

from abc import abstractmethod


class FilterFactory:
    available_filters = {}

    @classmethod
    def register(cls, filter_: type):
        # if not issubclass(filter_, Filter):
        #     raise InvalidFilterError(f'Invalid filter: {filter_}')
        cls.available_filters[filter_.name] = filter_
        setattr(cls, filter_.name, filter_)

    def __new__(cls, name, *args, **kwargs):
        if name not in cls.available_filters:
            raise ValueError(f'Unknown filter: {name}')
        return cls.available_filters[name](*args, **kwargs)


class MetaFilter(type):
    def __new__(cls, name, bases, attrs):
        new_class = super().__new__(cls, name, bases, attrs)
        if not name.startswith('Filter') and name != 'BaseFilter':
            raise ValueError('Filter class names must start with "Filter"')
        new_class.name = name.split('Filter', maxsplit=1)[1].lower()
        if name != 'BaseFilter':
            FilterFactory.register(new_class)
        return new_class


class BaseFilter(metaclass=MetaFilter):
    """ Base class for filters. """
    @abstractmethod
    def filter(self, value: str) -> str:
        raise NotImplementedError("Filter.filter() must be implemented")

    def __init__(self, *args, **kwargs):
        ...

    def __repr__(self):
        return f'{self.__class__.__name__}'

    def __call__(self, value: str) -> str:
        return self.filter(value)


class FilterUpper(BaseFilter):
    def filter(self, value: str) -> str:
        return value.upper()


class FilterRegex(BaseFilter):
    def __init__(self, pattern: str, replace: str):
        self.pattern = re.compile(pattern)
        self.replace = replace
        
    def filter(self, value: str) -> str:
        return self.pattern.sub(value, self.replace)

现有实现的三个缺陷

  1. 无法确保传入register的过滤器是BaseFilter的子类——因为BaseFilter是在元类之后声明的,Python不支持C++式的前向声明,注释掉的类型检查代码无法生效;
  2. 必须特意排除抽象类BaseFilter,不让它被添加到available_filters中;
  3. 整体模式感觉有些怪异,不够直观。

后续扩展:参数提取与Schema验证

我的目标是利用FilterFactory.available_filters构建基于Voluptuous的JSON Schema验证器,确保只接受可用过滤器,并且能在程序运行期间多次创建和应用过滤器。我在元类中添加了参数提取和类型检查逻辑:

class MetaFilter(type):
    def __new__(cls, name, bases, attrs):
        ...
        new_class.__params__, new_class.__types__ = \
            cls.extract_parameters(new_class)
        return new_class
    
    @classmethod
    def extract_parameters(cls, new_class):
        """ Extract parameters from the class.
        Ensure that all the parameters are annotated."""
        params = dict(inspect.signature(new_class.__init__).parameters)
        for key in ['self', 'args', 'kwargs']:
            if key in params:
                del params[key]
        for param, value in params.items():
            if value.annotation is inspect.Parameter.empty:
                raise ValueError(
                    f'Filter {new_class.name} has an untyped parameter: {param}'
                )
    
        return (params.keys(), [p.annotation for p in params.values()])

之后就可以创建验证Schema并使用:

filters = {}
for filter_name, filter_class in FilterFactory.available_filters.items():
    filters[Optional(filter_name)] = All(
        ExactSequence(filter_class.__types__),
        lambda args: FilterFactory(filter_name, *args)
    )
schema = Schema({'filter': filters})

s = schema({
    'filter': {
        'regex': ['foo', 'bar']
    }
})

assert(s['filter']['regex'].filter('foo') == 'bar')

现在新增过滤器只需要在filters.py模块中添加对应的类即可,但我不确定这个实现是否符合Python禅道(Pythonic)?还有哪些更优的替代方案?


解答

关于是否符合Python禅道

你的元类实现核心思路是合理的——自动注册过滤器、无需手动维护列表,符合"简单胜于复杂"的原则,但确实存在几个不够Pythonic的点:

  • 元类属于进阶特性,除非必要,Python更倾向用直观的方式解决问题;
  • 依赖类名前缀(FilterXXX)识别过滤器属于"魔术行为",不够显式;
  • 前向声明问题导致无法做子类检查,破坏了代码健壮性。

整体来说,满足需求的核心逻辑没问题,但实现方式可以更优雅。

更优的替代方案

方案1:用装饰器替代元类实现自动注册

装饰器比元类更直观,完全解决原方案的三个缺陷:

from abc import ABC, abstractmethod
import inspect
import re

class FilterFactory:
    available_filters = {}

    @classmethod
    def register(cls, filter_name=None):
        def decorator(filter_cls):
            # 直接检查子类关系,解决原方案的第一个缺陷
            if not issubclass(filter_cls, BaseFilter):
                raise ValueError(f"{filter_cls.__name__} must inherit from BaseFilter")
            # 支持自定义名字,无指定则从类名生成
            name = filter_name or filter_cls.__name__.replace("Filter", "").lower()
            cls.available_filters[name] = filter_cls
            setattr(cls, name, filter_cls)
            # 提取参数信息并验证
            cls._extract_parameters(filter_cls)
            return filter_cls
        return decorator

    @classmethod
    def _extract_parameters(cls, filter_cls):
        params = dict(inspect.signature(filter_cls.__init__).parameters)
        for key in ['self', 'args', 'kwargs']:
            params.pop(key, None)
        for param, value in params.items():
            if value.annotation is inspect.Parameter.empty:
                raise ValueError(f'Filter {filter_cls.__name__} has an untyped parameter: {param}')
        filter_cls.__params__ = list(params.keys())
        filter_cls.__types__ = [p.annotation for p in params.values()]

    def __new__(cls, name, *args, **kwargs):
        if name not in cls.available_filters:
            raise ValueError(f'Unknown filter: {name}')
        return cls.available_filters[name](*args, **kwargs)

class BaseFilter(ABC):
    """ Base class for filters. """
    @abstractmethod
    def filter(self, value: str) -> str:
        raise NotImplementedError("Filter.filter() must be implemented")

    def __init__(self, *args, **kwargs):
        ...

    def __repr__(self):
        return f'{self.__class__.__name__}'

    def __call__(self, value: str) -> str:
        return self.filter(value)

# 用装饰器注册过滤器
@FilterFactory.register()
class FilterUpper(BaseFilter):
    def filter(self, value: str) -> str:
        return value.upper()

@FilterFactory.register()
class FilterRegex(BaseFilter):
    def __init__(self, pattern: str, replace: str):
        self.pattern = re.compile(pattern)
        self.replace = replace
        
    def filter(self, value: str) -> str:
        return self.pattern.sub(self.replace, value)  # 修正原代码参数顺序错误

优势:

  • 显式注册逻辑,新人更容易理解;
  • 支持自定义过滤器名字,灵活性更高;
  • 无需特意排除BaseFilter,逻辑更简洁。

方案2:利用__subclasses__()动态获取子类

如果不需要模块加载时立即注册,可以用BaseFilter.__subclasses__()自动获取所有子类:

class FilterFactory:
    @classmethod
    def available_filters(cls):
        fmap = {}
        for subclass in BaseFilter.__subclasses__():
            # 跳过抽象子类
            if inspect.isabstract(subclass):
                continue
            name = subclass.__name__.replace("Filter", "").lower()
            fmap[name] = subclass
            cls._extract_parameters(subclass)
        return fmap

    # _extract_parameters和__new__方法同方案1
    # BaseFilter定义同前,无需装饰器

优势:

  • 完全无注册逻辑,代码极简;
  • 自动获取所有子类,无需手动维护。

局限性:

  • 子类必须被导入后才会出现在__subclasses__()结果中;
  • 无法自定义过滤器名字,只能依赖类名生成。

方案3:导入时扫描指定模块的过滤器

如果过滤器都集中在某个模块(比如filters.py),可以在模块导入时自动扫描注册:

# filters.py文件内
import inspect
import sys
from abc import ABC, abstractmethod

class BaseFilter(ABC):
    # 定义同前

class FilterFactory:
    available_filters = {}

    @classmethod
    def load_filters(cls, module):
        for name, member in inspect.getmembers(module):
            if inspect.isclass(member) and issubclass(member, BaseFilter) and member != BaseFilter:
                filter_name = name.replace("Filter", "").lower()
                cls.available_filters[filter_name] = member
                cls._extract_parameters(member)

    # _extract_parameters和__new__方法同方案1

# 模块导入时自动加载当前模块的过滤器
FilterFactory.load_filters(sys.modules[__name__])

优势:

  • 完全自动化,新增过滤器只需添加类;
  • 明确限制扫描范围,避免意外注册。

局限性:

  • 扫描逻辑相对隐蔽,新人需要时间理解;
  • 依赖类名生成过滤器名字,灵活性稍差。

总结

如果追求代码的直观性和灵活性,**方案1(装饰器)**是最优选择,完全符合Python"显式胜于隐式"的禅道;如果过滤器都集中在同一模块且追求极致简洁,**方案3(模块扫描)**也很合适;原元类方案虽然能工作,但确实不如装饰器直观优雅。

内容的提问来源于stack exchange,提问作者nowox

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.11 20:15:16