From 65310bcf2a1a366d2eba3a41f74e2fbc030ab077 Mon Sep 17 00:00:00 2001 From: m-paz Date: Wed, 7 Jul 2021 12:05:13 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Fixed=20flow=5Fof=5Fflows=20proj?= =?UTF-8?q?ect=5Fname=20issue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- viadot/flows/flow_of_flows.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/viadot/flows/flow_of_flows.py b/viadot/flows/flow_of_flows.py index 54af9e1ab..c6bf4b1a0 100644 --- a/viadot/flows/flow_of_flows.py +++ b/viadot/flows/flow_of_flows.py @@ -11,6 +11,7 @@ class Pipeline(Flow): def __init__( self, name: str, + project_name: str, extract_flows_names: List[str], transform_flow_name: str, *args: List[any], @@ -18,11 +19,14 @@ def __init__( ): self.extract_flows_names = extract_flows_names self.transform_flow_name = transform_flow_name + self.project_name = project_name super().__init__(*args, name=name, **kwargs) self.gen_flow() def gen_start_flow_run_task(self, flow_name: str, flow: Flow = None) -> Task: - t = start_flow_run_task.bind(flow_name=flow_name, flow=flow) + t = start_flow_run_task.bind( + flow_name=flow_name, project_name=self.project_name, flow=flow + ) return t def gen_flow(self) -> Flow: @@ -30,6 +34,8 @@ def gen_flow(self) -> Flow: self.gen_start_flow_run_task, self.extract_flows_names, flow=self ) transform_flow_run = start_flow_run_task_2.bind( - flow_name=self.transform_flow_name, flow=self + flow_name=self.transform_flow_name, + project_name=self.project_name, + flow=self, ) transform_flow_run.set_upstream(extract_flow_runs, flow=self)