如何让继承类调用类内定义函数，替代原本调用的全局函数？

你的核心问题是DailyParsingTaskDefinition的parsing_tasks方法硬编码调用了全局的count_identities_stage，导致无法切换到ConsolidatedDailyParsingDefinitionByMic类内的自定义实现。以下是两种可行的修改方案：

方案一：通过函数注入实现自定义逻辑 #

给DailyParsingTaskDefinition添加一个可配置的函数参数，允许创建实例时传入自定义的count_identities_stage实现：

1. 修改DailyParsingTaskDefinition类：

from typing import Callable
import dataclasses

@dataclasses.dataclass
class DailyParsingTaskDefinition(DailyDownloadDefinition):
    # 新增可选参数，默认使用全局函数
    count_identities_func: Callable = count_identities_stage

    def parsing_tasks(self):
        # 调用实例绑定的函数，而非全局函数
        self.count_identities = self.count_identities_func(self)

2. 在ConsolidatedDailyParsingDefinitionByMic的DAG方法中传入自定义函数：

@dataclasses.dataclass
class ConsolidatedDailyParsingDefinitionByMic(DailyProcessingParameters):

    def DAG(self):
        ...
        with DAG(dag_name, default_args=args, ...) as dag:

            for task_group_config in self.task_group_list:
                # 传入当前类的自定义方法
                task_group_instance = DailyParsingTaskDefinition(
                    **task_group_config,
                    count_identities_func=self.count_identities_stage
                )
                task_group_instance.parsing_tasks()

    # 修正方法定义，添加self参数（实例方法必须有）
    def count_identities_stage(self, dag_def):
        # 这里编写你的自定义逻辑
        ...
        return custom_count_identities

优点：无需子类化，灵活度高，适合临时替换单个函数的场景。

方案二：子类化DailyParsingTaskDefinition重写逻辑 #

通过继承DailyParsingTaskDefinition，重写parsing_tasks方法来调用自定义的count_identities_stage：

1. 创建自定义子类并关联外部实例：

import dataclasses

@dataclasses.dataclass
class CustomDailyParsingTaskDefinition(DailyParsingTaskDefinition):
    # 新增参数，保存外部类实例
    parent_def: ConsolidatedDailyParsingDefinitionByMic

    def parsing_tasks(self):
        # 调用外部类的自定义方法
        self.count_identities = self.parent_def.count_identities_stage(self)

2. 在ConsolidatedDailyParsingDefinitionByMic中使用自定义子类：

@dataclasses.dataclass
class ConsolidatedDailyParsingDefinitionByMic(DailyProcessingParameters):

    def DAG(self):
        ...
        with DAG(dag_name, default_args=args, ...) as dag:

            for task_group_config in self.task_group_list:
                task_group_instance = CustomDailyParsingTaskDefinition(
                    **task_group_config,
                    parent_def=self
                )
                task_group_instance.parsing_tasks()

    def count_identities_stage(self, dag_def):
        # 自定义逻辑
        ...
        return custom_count_identities

优点：符合面向对象的继承设计，适合需要扩展多个方法的场景。

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