> ## Documentation Index
> Fetch the complete documentation index at: https://docs.getcollate.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Advanced Usage

> Advanced patterns and configurations for Data Quality as Code

# Advanced Usage

This guide covers advanced patterns and configurations for Data Quality as Code, including loading tests from YAML files, customizing workflow configurations, and integrating with production systems.

## Loading Tests from YAML

Load test definitions from YAML workflow files to enable version-controlled test configurations:

### Basic YAML Loading

Use `TestRunner.from_yaml()` to load test definitions from a file or an inline YAML string.

```python theme={null}
from metadata.sdk.data_quality import TestRunner

# Load from YAML file
runner = TestRunner.from_yaml(file_path="tests/customer_quality.yaml")

# Or from YAML string
yaml_config = """
source:
  type: TestSuite
  serviceName: local_postgres
  sourceConfig:
    config:
      type: TestSuite
      entityFullyQualifiedName: Postgres.warehouse.public.customers

processor:
  type: orm-test-runner
  config:
    testCases:
      - name: customer_email_not_null
        testDefinitionName: columnValuesToBeNotNull
        columnName: email
      - name: customer_id_unique
        testDefinitionName: columnValuesToBeUnique
        columnName: customer_id

workflowConfig:
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: openmetadata
    securityConfig:
      jwtToken: your-token-here
"""

runner = TestRunner.from_yaml(yaml_string=yaml_config)

# Run the loaded tests
results = runner.run()
```

### Using Collate Connection from YAML

By default, `from_yaml()` uses the connection configured via `configure()`. To use the connection from the YAML file:

```python theme={null}
runner = TestRunner.from_yaml(
    file_path="tests/config.yaml",
    use_connection_from_yaml=True
)
```

### YAML File Structure

A complete YAML configuration includes:

```yaml theme={null}
source:
  type: TestSuite
  serviceName: postgres_production
  sourceConfig:
    config:
      type: TestSuite
      entityFullyQualifiedName: Postgres.analytics.public.user_events

processor:
  type: orm-test-runner
  config:
    forceUpdate: false
    testCases:
      # Table-level tests
      - name: table_row_count_validation
        testDefinitionName: tableRowCountToBeBetween
        parameterValues:
          - name: minValue
            value: "10000"
          - name: maxValue
            value: "1000000"

      # Column-level tests
      - name: user_id_not_null
        testDefinitionName: columnValuesToBeNotNull
        columnName: user_id

      - name: event_timestamp_format
        testDefinitionName: columnValuesToMatchRegex
        columnName: event_timestamp
        parameterValues:
          - name: regex
            value: "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z$"

workflowConfig:
  loggerLevel: INFO
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: openmetadata
    securityConfig:
      jwtToken: ${OPENMETADATA_JWT_TOKEN}
```

## Advanced TestRunner Configuration

Use these options to control runner behavior and inspect loaded test definitions.

### Customizing Workflow Behavior

Use the `setup()` method to control logging, error handling, and success thresholds.

```python theme={null}
from metadata.sdk.data_quality import TestRunner
from metadata.generated.schema.metadataIngestion.workflow import LogLevels

runner = TestRunner.for_table("BigQuery.analytics.events.user_sessions")

# Configure detailed settings
runner.setup(
    force_test_update=True,           # Update existing test definitions
    log_level=LogLevels.DEBUG,        # Enable debug logging
    raise_on_error=False,             # Continue on errors
    success_threshold=95,             # Require 95% success rate
    enable_streamable_logs=True       # Stream logs in real-time
)

# Add tests and run
runner.add_test(TableRowCountToBeBetween(min_count=1000))
results = runner.run()
```

### Accessing Test Definitions

Inspect configured tests before running:

```python theme={null}
runner = TestRunner.for_table("MySQL.ecommerce.public.orders")
runner.add_tests(
    TableRowCountToBeBetween(min_count=100),
    ColumnValuesToBeNotNull(column="order_id")
)

# Access test definitions
for test_def in runner.test_definitions:
    print(f"Test: {test_def.testDefinitionName}")
    print(f"Parameters: {test_def.parameterValues}")
```

## Publishing Results to Collate

Results can be published back to Collate for tracking, alerting, and visualization:

### DataFrame Validation Results

After validating a DataFrame, call `publish()` to send results to Collate.

```python theme={null}
from metadata.sdk.data_quality.dataframes import DataFrameValidator

validator = DataFrameValidator()
validator.add_openmetadata_table_tests("Postgres.staging.public.customers")

result = validator.validate(df)

# Publish results to Collate
result.publish("Postgres.staging.public.customers")
```

### Benefits of Publishing Results

Publishing results to Collate enables the following.

* **Historical tracking**: View trends over time.
* **Alerting**: Trigger notifications on failures.
* **Dashboards**: Centralized data quality monitoring.
* **Collaboration**: Share results across teams.
* **Compliance**: Maintain audit trails.

## Error Handling and Retries

Implement robust error handling:

```python theme={null}
import time
from metadata.sdk.data_quality import TestRunner

def run_with_retry(table_fqn, max_retries=3, backoff=2):
    """Run tests with exponential backoff retry"""
    for attempt in range(max_retries):
        try:
            runner = TestRunner.for_table(table_fqn)
            results = runner.run()
            return results

        except ConnectionError as e:
            if attempt < max_retries - 1:
                wait_time = backoff ** attempt
                print(f"Connection failed, retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                print(f"Failed after {max_retries} attempts")
                raise

        except ValueError as e:
            print(f"Configuration error: {e}")
            raise  # Don't retry configuration errors

        except Exception as e:
            print(f"Unexpected error: {e}")
            raise

# Usage
results = run_with_retry("Postgres.warehouse.public.customers")
```

## Dynamic Test Generation

Generate tests programmatically based on metadata:

```python theme={null}
from metadata.generated.schema.entity.data.table import Table
from metadata.sdk import configure, client
from metadata.sdk.data_quality import (
    TestRunner,
    ColumnValuesToBeNotNull,
    ColumnValuesToBeUnique
)

configure(host="http://localhost:8585/api", jwt_token="token")

# Get table metadata
om_client = client()
table = om_client.ometa.get_by_name(
    entity=Table,
    fqn="Postgres.warehouse.public.customers"
)

# Generate tests based on column types
runner = TestRunner.for_table(table.fullyQualifiedName.root)

for column in table.columns:
    # Add NOT NULL tests for required columns
    if column.constraint == "NOT NULL":
        runner.add_test(ColumnValuesToBeNotNull(column=column.name.root))

    # Add UNIQUE tests for primary keys
    if column.constraint == "PRIMARY KEY":
        runner.add_test(ColumnValuesToBeUnique(column=column.name.root))

results = runner.run()
```

## Multi-Table Validation

Validate multiple tables in a workflow:

```python theme={null}
from metadata.sdk.data_quality import TestRunner, TableRowCountToBeBetween

tables_to_validate = {
    "Postgres.warehouse.public.customers": {"min_rows": 10000},
    "Postgres.warehouse.public.orders": {"min_rows": 50000},
    "Postgres.warehouse.public.products": {"min_rows": 1000}
}

validation_results = {}

for table_fqn, config in tables_to_validate.items():
    runner = TestRunner.for_table(table_fqn)
    runner.add_test(TableRowCountToBeBetween(min_count=config["min_rows"]))

    results = runner.run()
    validation_results[table_fqn] = {
        "passed": all(r.testCaseResult.testCaseStatus == "Success" for r in results),
        "details": results
    }

# Generate summary
total_tables = len(validation_results)
passed_tables = sum(1 for v in validation_results.values() if v["passed"])

print(f"\n{'='*60}")
print(f"Validation Summary: {passed_tables}/{total_tables} tables passed")
print(f"{'='*60}")

for table_fqn, result in validation_results.items():
    status = "✓" if result["passed"] else "✗"
    print(f"{status} {table_fqn}")
```

## Best Practices Summary

Follow these guidelines when building production data quality workflows.

* **Version-control test configurations**: Store YAML test files in your git repository alongside your pipeline code. This keeps test definitions auditable, reviewable, and aligned with the code they validate.

* **Use environment variables for credentials**: Never hardcode JWT tokens or connection strings in code. Read them from environment variables such as `OPENMETADATA_JWT_TOKEN` and `OPENMETADATA_HOST` so credentials stay out of source control.

* **Implement retries for transient failures**: Network timeouts and temporary service unavailability are expected in production. Wrap `TestRunner.run()` calls in a retry loop with exponential backoff, and distinguish retryable errors (connection failures) from non-retryable ones (configuration errors).

* **Publish results after every run**: Calling `result.publish()` sends test outcomes to Collate, where they appear in dashboards and trigger configured alerts. Without publishing, test runs are invisible to data stewards and stakeholders.

* **Monitor test execution time**: Track how long test runs take over time. A sudden increase can indicate a growing dataset, a degraded connection, or a misconfigured test.

* **Handle errors explicitly**: Catch and log specific exception types rather than letting all exceptions fail silently. Distinguish between configuration errors, which should stop the pipeline, and transient errors, which may be retried.

* **Use descriptive test names and descriptions**: A test named `customer_email_not_null` with a description such as "Ensures all customer records have a valid email before loading to the warehouse" is far easier to triage than a generic name.

* **Validate incrementally, not just at the end**: Run column-level checks with `DataFrameValidator` during your ETL pipeline and table-level checks with `TestRunner` after loading. Catching issues early prevents bad data from reaching your warehouse.

* **Separate test ownership from execution**: Let data stewards define and maintain test criteria in the Collate UI. Engineers load and run those tests in pipelines using `runner.run()` without re-defining them in code. This keeps business logic out of pipeline scripts.

* **Confirm test definitions work before deploying**: Before adding a new test to a production pipeline, run it against a known-good dataset and a known-bad dataset to confirm it passes and fails as expected.

## Next Steps

Once you've completed advanced configuration, explore these related guides.

* Review the [Test Definitions Reference](/ai-2-0/how-to-guides/data-quality-observability/quality/data-quality-as-code/test-definitions).
* Learn about [TestRunner](/ai-2-0/how-to-guides/data-quality-observability/quality/data-quality-as-code/test-runner).
* Explore [DataFrame Validation](/ai-2-0/how-to-guides/data-quality-observability/quality/data-quality-as-code/dataframe-validation).
* Return to the [Data Quality as Code Overview](/ai-2-0/how-to-guides/data-quality-observability/quality/data-quality-as-code).
* Explore the [Examples and Tutorials](https://github.com/open-metadata/OpenMetadata/tree/main/examples/python-sdk/data-quality/README.md) for additional patterns.