相关疑难解决方法(0)

假设我有两个不同的类实现

class ParentA:
    def initialize(self):
        pass

    def some_event(self):
        pass

    def order(self, value):
        # handle order in some way for Parent A


class ParentB:
    def initialize(self):
        pass

    def some_event(self):
        pass

    def order(self, value):
        # handle order in another for Parent B

我怎样才能动态地让一些第三类继承ParentA或ParentB基于这样的东西？

class MyCode:
    def initialize(self):
        self.initial_value = 1

    def some_event(self):
        # handle event
        order(self.initial_value)


# let MyCode inherit from ParentA and run
run(my_code, ParentA)

我想将包含 google 存储中文件名的字符串列表传递给 XCom。稍后由 GoogleCloudStorageToBigQueryOperator 任务获取。该source_objects场模板，使神社模板可以使用。不幸的是，Jinja 只能返回一个字符串，因此我无法在 XCom 中传递列表。

如何在 GoogleCloudStorageToBigQueryOperator 中使用 XCom 列表？

参考类似问题，通过使用provide_context解决： 将字符串列表作为 Airflow 中依赖任务的参数传递

我找到的最接近的解决方案是创建一个包装类并发送发布 xcom 的任务的 ID，如下所示：

@apply_defaults
def __init__(self, source_objects_task_id,
....
def execute(self, context):
    source_objects = context['ti']
          .xcom_pull(task_ids=self.source_objects_task_id)
    operator = GoogleCloudStorageToBigQueryOperator(
          source_objects=source_objects,
          dag=self.dag,
....
)

    operator.execute(context)