yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-15 12:37:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

MINOR - Add labels in OM Airflow DAGs (#16233)

Browse Source 
* MINOR - Add labels in OM Airflow DAGs

* update test
This commit is contained in:
Pere Miquel Brull 2024-05-13 12:45:20 +02:00 committed by GitHub
parent 4f9415fef6
commit db3bff19af
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 2 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
openmetadata-airflow-apis/openmetadata_managed_apis/workflows/ingestion/common.py
View File

@ -255,7 +255,9 @@ def build_dag_configs(ingestion_pipeline: IngestionPipeline) -> dict:
"schedule_interval": ingestion_pipeline.airflowConfig.scheduleInterval, "schedule_interval": ingestion_pipeline.airflowConfig.scheduleInterval,
"tags": [ "tags": [
"OpenMetadata", "OpenMetadata",
ingestion_pipeline.displayName or ingestion_pipeline.name.__root__,
ingestion_pipeline.pipelineType.value, ingestion_pipeline.pipelineType.value,
ingestion_pipeline.service.name,
], ],
} }

55
openmetadata-airflow-apis/tests/unit/ingestion_pipeline/test_workflow_creation.py
View File

@ -17,9 +17,6 @@ import uuid
from unittest import TestCase from unittest import TestCase
from unittest.mock import patch from unittest.mock import patch
from openmetadata_managed_apis.workflows.ingestion.application import (
build_application_workflow_config,
)
from openmetadata_managed_apis.workflows.ingestion.lineage import ( from openmetadata_managed_apis.workflows.ingestion.lineage import (
build_lineage_workflow_config, build_lineage_workflow_config,
) )
@ -36,12 +33,6 @@ from openmetadata_managed_apis.workflows.ingestion.usage import (
build_usage_workflow_config, build_usage_workflow_config,
) )
from metadata.generated.schema.entity.applications.configuration.applicationConfig import (
AppConfig,
)
from metadata.generated.schema.entity.applications.configuration.external.autoTaggerAppConfig import (
AutoTaggerAppConfig,
)
from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import ( from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
OpenMetadataConnection, OpenMetadataConnection,
) )
@ -51,10 +42,6 @@ from metadata.generated.schema.entity.services.ingestionPipelines.ingestionPipel
IngestionPipeline, IngestionPipeline,
PipelineType, PipelineType,
) )
from metadata.generated.schema.metadataIngestion.applicationPipeline import (
ApplicationConfigType,
ApplicationPipeline,
)
from metadata.generated.schema.metadataIngestion.databaseServiceMetadataPipeline import ( from metadata.generated.schema.metadataIngestion.databaseServiceMetadataPipeline import (
DatabaseServiceMetadataPipeline, DatabaseServiceMetadataPipeline,
) )
@ -81,7 +68,6 @@ from metadata.generated.schema.type.entityReference import EntityReference
from metadata.ingestion.api.parser import parse_workflow_config_gracefully from metadata.ingestion.api.parser import parse_workflow_config_gracefully
from metadata.ingestion.models.encoders import show_secrets_encoder from metadata.ingestion.models.encoders import show_secrets_encoder
from metadata.ingestion.ometa.ometa_api import OpenMetadata from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.workflow.application import ApplicationWorkflow
from metadata.workflow.data_quality import TestSuiteWorkflow from metadata.workflow.data_quality import TestSuiteWorkflow
from metadata.workflow.metadata import MetadataWorkflow from metadata.workflow.metadata import MetadataWorkflow
from metadata.workflow.profiler import ProfilerWorkflow from metadata.workflow.profiler import ProfilerWorkflow
@ -358,44 +344,3 @@ class OMetaServiceTest(TestCase):
config = json.loads(workflow_config.json(encoder=show_secrets_encoder)) config = json.loads(workflow_config.json(encoder=show_secrets_encoder))
parse_workflow_config_gracefully(config) parse_workflow_config_gracefully(config)
@patch.object(
ApplicationWorkflow,
"set_ingestion_pipeline_status",
mock_set_ingestion_pipeline_status,
)
def test_application_workflow(self):
"""
Validate that the ingestionPipeline can be parsed
and properly load an Application Workflow
"""
ingestion_pipeline = IngestionPipeline(
id=uuid.uuid4(),
name="test_auto_tagger_application",
pipelineType=PipelineType.application,
fullyQualifiedName="OpenMetadata.test_auto_tagger_application",
sourceConfig=SourceConfig(
config=ApplicationPipeline(
type=ApplicationConfigType.Application,
appConfig=AppConfig(
__root__=AutoTaggerAppConfig(confidenceLevel=80)
),
sourcePythonClass="metadata.applications.auto_tagger.AutoTaggerApp",
)
),
openMetadataServerConnection=self.server_config,
airflowConfig=AirflowConfig(
startDate="2022-06-10T15:06:47+00:00",
),
service=EntityReference(
id=uuid.uuid4(),
type="metadata",
name="OpenMetadata",
),
)
workflow_config = build_application_workflow_config(ingestion_pipeline)
config = json.loads(workflow_config.json(encoder=show_secrets_encoder))
ApplicationWorkflow.create(config)