fix for dbt hanging while running

valmi-io · Jun 17, 2023 · 7bebb95 · 7bebb95
1 parent ae2c6ad
commit 7bebb95
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 3 deletions.
diff --git a/valmi-integrations/connectors/source-postgres/valmi_dbt/dbt_airbyte_adapter.py b/valmi-integrations/connectors/source-postgres/valmi_dbt/dbt_airbyte_adapter.py
@@ -217,6 +217,11 @@ def execute_dbt(self, logger: AirbyteLogger):
             stderr=subprocess.PIPE,
             stdout=subprocess.PIPE,
         )
+
+        logs = []
+        for line in iter(proc.stdout.readline, b''):
+            logs.append(line)
+        proc.stdout.close()
 
         while True:
             try:
@@ -226,7 +231,6 @@ def execute_dbt(self, logger: AirbyteLogger):
                 continue
             break
 
-        logs = proc.stdout.readlines()
         lastError = None
         for log in logs:
             j = json.loads(log.decode())

diff --git a/valmi-integrations/connectors/source-redshift/valmi_dbt/dbt_airbyte_adapter.py b/valmi-integrations/connectors/source-redshift/valmi_dbt/dbt_airbyte_adapter.py
@@ -217,6 +217,11 @@ def execute_dbt(self, logger: AirbyteLogger):
             stdout=subprocess.PIPE,
         )
 
+        logs = []
+        for line in iter(proc.stdout.readline, b''):
+            logs.append(line)
+        proc.stdout.close()
+
         while True:
             try:
                 proc.wait(timeout=2)
@@ -225,7 +230,6 @@ def execute_dbt(self, logger: AirbyteLogger):
                 continue
             break
 
-        logs = proc.stdout.readlines()
         lastError = None
         for log in logs:
             j = json.loads(log.decode())

diff --git a/valmi-integrations/connectors/source-snowflake/valmi_dbt/dbt_airbyte_adapter.py b/valmi-integrations/connectors/source-snowflake/valmi_dbt/dbt_airbyte_adapter.py
@@ -242,6 +242,11 @@ def execute_dbt(self, logger: AirbyteLogger):
             stderr=subprocess.PIPE,
             stdout=subprocess.PIPE,
         )
+
+        logs = []
+        for line in iter(proc.stdout.readline, b''):
+            logs.append(line)
+        proc.stdout.close()
 
         while True:
             try:
@@ -251,7 +256,6 @@ def execute_dbt(self, logger: AirbyteLogger):
                 continue
             break
 
-        logs = proc.stdout.readlines()
         lastError = None
         for log in logs:
             j = json.loads(log.decode())