DQ base workflow

ingestion/operators/docker/main.py

@@ -15,13 +15,13 @@
 
 import yaml
 
-from metadata.data_quality.api.workflow import TestSuiteWorkflow
 from metadata.generated.schema.entity.services.ingestionPipelines.ingestionPipeline import (
  PipelineType,
 )
 from metadata.generated.schema.metadataIngestion.workflow import LogLevels
 from metadata.utils.logger import set_loggers_level
 from metadata.workflow.data_insight import DataInsightWorkflow
+from metadata.workflow.data_quality import TestSuiteWorkflow
 from metadata.workflow.metadata import MetadataWorkflow
 from metadata.workflow.profiler import ProfilerWorkflow
 from metadata.workflow.usage import UsageWorkflow

ingestion/src/metadata/cli/dataquality.py

@@ -17,12 +17,12 @@
 import traceback
 
 from metadata.config.common import load_config_file
-from metadata.data_quality.api.workflow import TestSuiteWorkflow
 from metadata.utils.logger import cli_logger
+from metadata.workflow.data_quality import TestSuiteWorkflow
 from metadata.workflow.workflow_output_handler import (
  WorkflowType,
  print_init_error,
- print_test_suite_status,
+ print_status,
 )
 
 logger = cli_logger()
@@ -48,5 +48,5 @@ def run_test(config_path: str) -> None:
 
  workflow.execute()
  workflow.stop()
- print_test_suite_status(workflow)
+ print_status(workflow)
  workflow.raise_from_status()

ingestion/src/metadata/data_quality/api/models.py

@@ -18,8 +18,13 @@
 
 from typing import List, Optional
 
+from pydantic import BaseModel, Field
+
 from metadata.config.common import ConfigModel
-from metadata.generated.schema.tests.testCase import TestCaseParameterValue
+from metadata.generated.schema.api.tests.createTestSuite import CreateTestSuiteRequest
+from metadata.generated.schema.entity.data.table import Table
+from metadata.generated.schema.tests.basic import TestCaseResult
+from metadata.generated.schema.tests.testCase import TestCase, TestCaseParameterValue
 
 
 class TestCaseDefinition(ConfigModel):
@@ -38,3 +43,27 @@ class TestSuiteProcessorConfig(ConfigModel):
 
  testCases: Optional[List[TestCaseDefinition]] = None
  forceUpdate: Optional[bool] = False
+
+
+class TestCaseResultResponse(BaseModel):
+ testCaseResult: TestCaseResult
+ testCase: TestCase
+
+
+class TableAndTests(BaseModel):
+ """Source response bringing together the table and test cases"""
+
+ table: Table = Field(None, description="Table being processed by the DQ workflow")
+ service_type: str = Field(..., description="Service type the table belongs to")
+ test_cases: Optional[List[TestCase]] = Field(
+ None, description="Test Cases already existing in the Test Suite, if any"
+ )
+ executable_test_suite: Optional[CreateTestSuiteRequest] = Field(
+ None, description="If no executable test suite is found, we'll create one"
+ )
+
+
+class TestCaseResults(BaseModel):
+ """Processor response with a list of computed Test Case Results"""
+
+ test_results: Optional[List[TestCaseResultResponse]]