# Copyright 2025 Collate # Licensed under the Collate Community License, Version 1.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ Auto Classification DAG function builder """ import json from airflow import DAG from openmetadata_managed_apis.utils.logger import set_operator_logger from openmetadata_managed_apis.workflows.ingestion.common import ( build_dag, build_source, execute_workflow, ) from metadata.generated.schema.entity.services.ingestionPipelines.ingestionPipeline import ( IngestionPipeline, ) from metadata.generated.schema.metadataIngestion.workflow import ( LogLevels, OpenMetadataWorkflowConfig, Processor, Sink, WorkflowConfig, ) from metadata.workflow.classification import AutoClassificationWorkflow def auto_classification_workflow( workflow_config: OpenMetadataWorkflowConfig, ): """ Task that creates and runs the auto classification workflow. The workflow_config gets cooked form the incoming ingestionPipeline. This is the callable used to create the PythonOperator """ set_operator_logger(workflow_config) config = json.loads( workflow_config.model_dump_json(exclude_defaults=False, mask_secrets=False) ) workflow = AutoClassificationWorkflow.create(config) execute_workflow(workflow, workflow_config) def build_auto_classification_workflow_config( ingestion_pipeline: IngestionPipeline, ) -> OpenMetadataWorkflowConfig: """ Given an airflow_pipeline, prepare the workflow config JSON """ workflow_config = OpenMetadataWorkflowConfig( source=build_source(ingestion_pipeline), sink=Sink( type="metadata-rest", config={}, ), processor=Processor( type="orm-profiler", config={}, ), workflowConfig=WorkflowConfig( loggerLevel=ingestion_pipeline.loggerLevel or LogLevels.INFO, openMetadataServerConfig=ingestion_pipeline.openMetadataServerConnection, ), ingestionPipelineFQN=ingestion_pipeline.fullyQualifiedName.root, ) return workflow_config def build_auto_classification_dag(ingestion_pipeline: IngestionPipeline) -> DAG: """ Build a simple metadata workflow DAG """ workflow_config = build_auto_classification_workflow_config(ingestion_pipeline) dag = build_dag( task_name="auto_classification_task", ingestion_pipeline=ingestion_pipeline, workflow_config=workflow_config, workflow_fn=auto_classification_workflow, ) return dag