如何在Python脚本中动态传入函数参数实现Kafka消息过滤

我有一个Python脚本，功能是从指定Kafka主题读取消息，经过自定义过滤后将符合条件的消息发送到另一个Kafka主题。目前脚本用argparse接收--source_topic和--target_topic两个参数，核心逻辑伪代码如下：

for each message in source_topic:
    is_fit = check_if_message_fits_target_topic(message)
    if is_fit:
        produce(target_topic, message)

运行命令示例：

python3 my_script.py --source_topic someSourceTopic --target_topic someTargetTopic

现在我希望让check_if_message_fits_target_topic函数支持动态传入自定义逻辑，这样同一脚本可以根据不同需求使用不同的过滤规则。比如现有过滤逻辑是判断事件名称：

def check_if_message_fits_target_topic(message):
    values = message.value
    if values['event_name'] == 'some_event_name':
        return True
    return False

还需要支持其他逻辑，比如判断创建时间是否在昨天之后：

def check_if_message_fits_target_topic(message):
    values = message.value
    yesterday = datetime.date.today() - datetime.timedelta(days=1)

    if values['created_at'] > yesterday:
        return True
    return False

只要过滤函数返回布尔值True/False即可，请问如何基于现有的argparse参数管理，实现这种动态传入过滤逻辑的功能？

方案1：通过模块导入加载过滤函数（生产环境首选） #

这是最规范、易维护的方式，把过滤逻辑单独写成Python模块，脚本通过参数指定模块路径和函数名来加载。

步骤： #

1. 编写独立的过滤模块，比如filters.py，存放多个过滤函数：

# filters.py
import datetime

def filter_by_event_name(message):
    values = message.value
    return values['event_name'] == 'some_event_name'

def filter_by_yesterday(message):
    values = message.value
    yesterday = datetime.date.today() - datetime.timedelta(days=1)
    return values['created_at'] > yesterday

2. 修改脚本的argparse配置，新增--filter参数用于指定函数路径：

import argparse
import importlib

def load_filter_function(filter_path):
    # 拆分模块路径和函数名，比如"filters.filter_by_event_name"拆分为模块名+函数名
    module_name, func_name = filter_path.rsplit('.', 1)
    module = importlib.import_module(module_name)
    return getattr(module, func_name)

# 解析参数
parser = argparse.ArgumentParser()
parser.add_argument('--source_topic', required=True)
parser.add_argument('--target_topic', required=True)
parser.add_argument('--filter', required=True, help='格式：模块名.函数名，比如filters.filter_by_event_name')
args = parser.parse_args()

# 加载过滤函数
filter_func = load_filter_function(args.filter)

# 核心逻辑
for message in source_topic:
    if filter_func(message):
        produce(args.target_topic, message)

3. 运行命令示例：

# 使用事件名称过滤
python3 my_script.py --source_topic someSourceTopic --target_topic someTargetTopic --filter filters.filter_by_event_name

# 使用时间过滤
python3 my_script.py --source_topic someSourceTopic --target_topic someTargetTopic --filter filters.filter_by_yesterday

方案2：通过命令行传入Python代码字符串（仅用于快速测试） #

如果只是临时测试简单逻辑，可以直接把过滤函数的代码字符串作为参数传入，用exec加载。

修改脚本： #

import argparse

def load_filter_from_code(code_str):
    # 创建临时命名空间执行代码，取出定义好的过滤函数
    namespace = {}
    exec(code_str, namespace)
    return namespace['filter_func']

parser = argparse.ArgumentParser()
parser.add_argument('--source_topic', required=True)
parser.add_argument('--target_topic', required=True)
parser.add_argument('--filter-code', required=True, help='包含filter_func函数的Python代码字符串')
args = parser.parse_args()

filter_func = load_filter_from_code(args.filter_code)

# 核心逻辑
for message in source_topic:
    if filter_func(message):
        produce(args.target_topic, message)

运行命令示例： #

python3 my_script.py --source_topic someSourceTopic --target_topic someTargetTopic \
--filter-code "def filter_func(message): values = message.value; return values['event_name'] == 'some_event_name'"

⚠️ 注意：这种方式存在安全风险，不要在生产环境使用，传入的代码会被直接执行。

方案3：注册器模式（适合固定多场景的生产环境） #

如果你的过滤逻辑是预定义的几种，可以在脚本里注册这些函数，通过参数指定函数别名来选择。

修改脚本： #

import argparse
import datetime

# 注册过滤函数的字典
filter_registry = {}

def register_filter(name):
    def decorator(func):
        filter_registry[name] = func
        return func
    return decorator

# 定义并注册过滤函数
@register_filter('event_name')
def filter_by_event_name(message):
    values = message.value
    return values['event_name'] == 'some_event_name'

@register_filter('yesterday')
def filter_by_yesterday(message):
    values = message.value
    yesterday = datetime.date.today() - datetime.timedelta(days=1)
    return values['created_at'] > yesterday

parser = argparse.ArgumentParser()
parser.add_argument('--source_topic', required=True)
parser.add_argument('--target_topic', required=True)
parser.add_argument('--filter', required=True, choices=filter_registry.keys(), help='可选过滤规则：event_name, yesterday')
args = parser.parse_args()

filter_func = filter_registry[args.filter]

# 核心逻辑
for message in source_topic:
    if filter_func(message):
        produce(args.target_topic, message)

运行命令示例： #

python3 my_script.py --source_topic someSourceTopic --target_topic someTargetTopic --filter event_name

这种方式安全可控，新增过滤规则只需要添加带装饰器的函数即可。

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