Improve airflow operator (#10)

* Improve airflow OpenDbtExecutorOperator * Improve airflow OpenDbtExecutorOperator
memiiso · Jul 10, 2024 · 6e777b0 · 6e777b0
1 parent 99de09a
commit 6e777b0
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/opendbt/airflow/__init__.py b/opendbt/airflow/__init__.py
@@ -21,13 +21,16 @@ def __init__(self,
                  profiles_dir: Path = None,
                  select: str = None,
                  args: list = None,
+                 # without using subprocess airflow randomly gets deadlock
+                 use_subprocess: bool = True,
                  execution_timeout=timedelta(minutes=60), **kwargs) -> None:
         super().__init__(execution_timeout=execution_timeout, **kwargs)
 
         self.project_dir: Path = project_dir
         self.command = command
         self.profiles_dir: Path = profiles_dir
         self.target = target
+        self.use_subprocess = use_subprocess
         self.args = args if args else []
 
         if select:
@@ -46,7 +49,7 @@ def execute(self, context):
         runner = opendbt.OpenDbtProject(project_dir=self.project_dir,
                                         profiles_dir=self.profiles_dir,
                                         target=self.target)
-        runner.run(command=self.command, args=self.args, use_subprocess=True)
+        runner.run(command=self.command, args=self.args, use_subprocess=self.use_subprocess)
 
 
 class OpenDbtAirflowProject(opendbt.OpenDbtProject):