yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-11-01 11:09:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix #5702: Airbyte Pipeline Status Fix (#5726)

Browse Source
This commit is contained in:
Mayur Singal 2022-06-29 14:03:25 +05:30 committed by GitHub
parent 11dc2e4918
commit 81047c4d18
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 20 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
ingestion/src/metadata/ingestion/source/pipeline/airbyte.py
View File

@ -11,6 +11,7 @@
"""
Airbyte source to extract metadata
"""
import traceback
from dataclasses import dataclass, field
from typing import Iterable, List, Optional
@ -122,14 +123,11 @@ class AirbyteSource(Source[CreatePipelineRequest]):
"""
return [
Task(
name=job["job"]["id"],
displayName=job["job"]["id"],
name=connection["connectionId"],
displayName=connection["name"],
description="",
taskUrl=f"{connection_url}/status",
taskType=job["job"]["configType"],
)
for job in self.client.list_jobs(connection.get("connectionId"))
if job
]
def fetch_pipeline(
@ -158,25 +156,27 @@ class AirbyteSource(Source[CreatePipelineRequest]):
Method to get task & pipeline status
"""
for job in self.client.list_jobs(connection.get("connectionId")):
task_status = [
TaskStatus(
name=str(job["job"]["id"]),
if not job or job.get("attempts"):
continue
for attempt in job["attempts"]:
task_status = [
TaskStatus(
name=str(connection.get("connectionId")),
executionStatus=STATUS_MAP.get(
attempt["status"].lower(), StatusType.Pending
).value,
)
]
pipeline_status = PipelineStatus(
executionStatus=STATUS_MAP.get(
attempt["status"].lower(), StatusType.Pending
).value,
taskStatus=task_status,
executionDate=attempt["createdAt"],
)
yield OMetaPipelineStatus(
pipeline_fqn=pipeline_fqn, pipeline_status=pipeline_status
)
for attempt in job["attempts"]
]
pipeline_status = PipelineStatus(
executionStatus=STATUS_MAP.get(
job["job"]["status"].lower(), StatusType.Pending
).value,
taskStatus=task_status,
executionDate=job["job"]["createdAt"],
)
yield OMetaPipelineStatus(
pipeline_fqn=pipeline_fqn, pipeline_status=pipeline_status
)
def fetch_lineage(
self, connection: dict, pipeline_entity: Pipeline