┌─────────────────────────────────────────────────────────────┐
│                 Data Quality Framework                       │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │   Profiler  │  │   Tests     │  │   Incident Manager  │  │
│  │  (Metrics)  │→ │ (Assertions)│→ │   (Resolution)      │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
│         ↓                ↓                    ↓              │
│  ┌─────────────────────────────────────────────────────────┐│
│  │                 Alerts & Notifications                  ││
│  └─────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────┘

source:
  type: profiler
  serviceName: my-database
  sourceConfig:
    config:
      type: Profiler
      generateSampleData: true
      profileSampleType: PERCENTAGE
      profileSample: 50  # Sample 50% of rows
      tableFilterPattern:
        includes:
          - "prod.*"
        excludes:
          - ".*_staging"
      columnFilterPattern:
        includes:
          - ".*"

processor:
  type: orm-profiler
  config: {}

sink:
  type: metadata-rest
  config: {}

workflowConfig:
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: openmetadata
    securityConfig:
      jwtToken: ${OM_JWT_TOKEN}

source:
  type: TestSuite
  serviceName: my-database
  sourceConfig:
    config:
      type: TestSuite
      entityFullyQualifiedName: my-database.schema.table

processor:
  type: orm-test-runner
  config:
    testCases:
      - name: orders_row_count_check
        testDefinitionName: tableRowCountToBeBetween
        parameterValues:
          - name: minValue
            value: 1000
          - name: maxValue
            value: 1000000

      - name: customer_id_not_null
        testDefinitionName: columnValuesToBeNotNull
        columnName: customer_id

      - name: status_in_valid_set
        testDefinitionName: columnValuesToBeInSet
        columnName: status
        parameterValues:
          - name: allowedValues
            value: "['pending', 'completed', 'cancelled']"

sink:
  type: metadata-rest
  config: {}

workflowConfig:
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: openmetadata
    securityConfig:
      jwtToken: ${OM_JWT_TOKEN}

testSuites:
  - name: orders_table_suite
    description: Quality tests for orders table
    testCases:
      - orders_row_count_check
      - customer_id_not_null
      - status_in_valid_set

- name: custom_business_rule
  testDefinitionName: tableCustomSQLQuery
  parameterValues:
    - name: sqlExpression
      value: "SELECT COUNT(*) FROM orders WHERE total < 0"
    - name: strategy
      value: "COUNT"
    - name: threshold
      value: 0

- name: quality_by_region
  testDefinitionName: columnValuesToBeBetween
  columnName: revenue
  parameterValues:
    - name: minValue
      value: 0
    - name: partitionColumnName
      value: region
    - name: partitionValues
      value: "['US', 'EU', 'APAC']"

from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.tests.testCase import TestCase
from metadata.generated.schema.tests.testDefinition import TestDefinition
from metadata.generated.schema.type.entityReference import EntityReference

# Initialize client
metadata = OpenMetadata(server_config)

# Get test definition
null_test = metadata.get_by_name(
    entity=TestDefinition,
    fqn="columnValuesToBeNotNull",
)

# Create test case
test_case = TestCase(
    name="customer_id_not_null",
    testDefinition=EntityReference(
        id=null_test.id,
        type="testDefinition",
    ),
    entityLink="<#E::table::my-db.schema.orders::columns::customer_id>",
    parameterValues=[],
)

created = metadata.create_or_update(test_case)
print(f"Created test: {created.name}")

from metadata.workflow.data_quality import TestSuiteWorkflow

config = {
    "source": {
        "type": "TestSuite",
        "serviceName": "my-database",
        "sourceConfig": {
            "config": {
                "type": "TestSuite",
                "entityFullyQualifiedName": "my-database.schema.orders"
            }
        }
    },
    # ... rest of config
}

workflow = TestSuiteWorkflow.create(config)
workflow.execute()
workflow.print_status()

filters:
  - field: entityType
    condition: equals
    value: table
  - field: testResult
    condition: equals
    value: Failed
  - field: tier
    condition: in
    values: [Tier1, Tier2]

destinations:
  - type: Slack
    config:
      webhookUrl: https://hooks.slack.com/services/XXX/YYY/ZZZ
      channel: "#data-quality-alerts"

Test Failure
    ↓
New Incident Created
    ↓
Acknowledged (ack)
    ↓
Assigned to Owner
    ↓
Investigation
    ↓
Root Cause Documented
    ↓
Resolved (with reason)

1. Failure Notification → System alerts on test failure
2. Acknowledgment → Team member confirms awareness
3. Assignment → Routes to knowledgeable person
4. Status Updates → Assigned team communicates progress
5. Resolution → All stakeholders notified of fix

name: Data Quality Check
on:
  schedule:
    - cron: '0 6 * * *'
  workflow_dispatch:

jobs:
  quality-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'

      - name: Install dependencies
        run: pip install openmetadata-ingestion

      - name: Run quality tests
        env:
          OM_HOST: ${{ secrets.OM_HOST }}
          OM_TOKEN: ${{ secrets.OM_TOKEN }}
        run: |
          metadata_openmetadata test-suite run \
            --config quality-tests.yaml

      - name: Check results
        run: |
          # Fail pipeline if critical tests failed
          metadata_openmetadata test-suite status \
            --suite critical-tests \
            --fail-on-failure