Fix dbtcloud task/run date (#19265)

open-metadata · Jan 8, 2025 · 4a877de · 4a877de
1 parent f763575
commit 4a877de
Showing 1 changed file with 22 additions and 30 deletions.
diff --git a/ingestion/src/metadata/ingestion/source/pipeline/dbtcloud/metadata.py b/ingestion/src/metadata/ingestion/source/pipeline/dbtcloud/metadata.py
@@ -276,8 +276,16 @@ def yield_pipeline_status(
         Get Pipeline Status
         """
         try:
-            task_status = [
-                TaskStatus(
+
+            pipeline_fqn = fqn.build(
+                metadata=self.metadata,
+                entity_type=Pipeline,
+                service_name=self.context.get().pipeline_service,
+                pipeline_name=self.context.get().pipeline,
+            )
+
+            for task in self.client.get_runs(job_id=int(pipeline_details.id)) or []:
+                task_status = TaskStatus(
                     name=str(task.id),
                     executionStatus=STATUS_MAP.get(task.state, StatusType.Pending),
                     startTime=(
@@ -303,37 +311,21 @@ def yield_pipeline_status(
                         else None
                     ),
                 )
-                for task in self.client.get_runs(job_id=int(pipeline_details.id)) or []
-            ]
 
-            pipeline_status = PipelineStatus(
-                executionStatus=STATUS_MAP.get(
-                    pipeline_details.state, StatusType.Pending
-                ),
-                taskStatus=task_status,
-                timestamp=Timestamp(
-                    datetime_to_ts(
-                        datetime.strptime(
-                            pipeline_details.created_at, "%Y-%m-%d %H:%M:%S.%f%z"
-                        )
-                        if pipeline_details.created_at
-                        else None
-                    )
-                ),
-            )
+                pipeline_status = PipelineStatus(
+                    executionStatus=task_status.executionStatus,
+                    taskStatus=[task_status],
+                    timestamp=task_status.endTime
+                    if task_status.endTime
+                    else task_status.startTime,
+                )
 
-            pipeline_fqn = fqn.build(
-                metadata=self.metadata,
-                entity_type=Pipeline,
-                service_name=self.context.get().pipeline_service,
-                pipeline_name=self.context.get().pipeline,
-            )
-            yield Either(
-                right=OMetaPipelineStatus(
-                    pipeline_fqn=pipeline_fqn,
-                    pipeline_status=pipeline_status,
+                yield Either(
+                    right=OMetaPipelineStatus(
+                        pipeline_fqn=pipeline_fqn,
+                        pipeline_status=pipeline_status,
+                    )
                 )
-            )
 
         except Exception as exc:
             yield Either(