Getting Started with Data Quality as Code

This guide will help you install the OpenMetadata Python SDK and configure authentication to start running data quality tests programmatically.

Prerequisites

Before you begin, ensure you have:

Python 3.10 or higher installed
pip package manager
Access to an OpenMetadata instance (version 2.0.3 or later)
A JWT token for authentication (see Authentication below)

Installation

The examples below target an OpenMetadata 2.0.3 server and allow ingestion package patch releases within 2.0.3. For another server version, use the matching ingestion package version. Install the necessary connector extras for your use case:

Basic Installation

pip install "openmetadata-ingestion~=2.0.3.0"

Installation with Database Connectors

Install additional dependencies based on the databases you’ll be testing:

# For PostgreSQL
pip install "openmetadata-ingestion[postgres]~=2.0.3.0"

# For MySQL
pip install "openmetadata-ingestion[mysql]~=2.0.3.0"

# For BigQuery
pip install "openmetadata-ingestion[bigquery]~=2.0.3.0"

# For multiple databases
pip install "openmetadata-ingestion[postgres,mysql,bigquery]~=2.0.3.0"

Installation with DataFrame Support

If you plan to use DataFrame validation features:

pip install "openmetadata-ingestion[pandas]~=2.0.3.0"

Installation with Multiple Features

Combine multiple extras as needed:

# For DataFrame validation with Postgres support
pip install "openmetadata-ingestion[pandas,postgres]~=2.0.3.0"

# For comprehensive ETL support
pip install "openmetadata-ingestion[pandas,postgres,pyarrow]~=2.0.3.0"

Authentication

Data Quality as Code requires authentication with your OpenMetadata instance. The SDK supports JWT token authentication.

Getting a JWT Token

You can obtain a JWT token in two ways:

Note: The Bots tile under Settings is only visible to users with Admin privileges. If you don’t see it, ask your organization’s OpenMetadata Admin to generate a bot token for you or grant you Admin access.

Option 1: Using an Existing Bot Token

OpenMetadata provides pre-configured bots like the ingestion-bot:

Log in to your OpenMetadata instance
Navigate to Settings > Bots
Find the ingestion-bot (or create a new bot)
Copy the JWT token

Option 2: Creating a Custom Bot

For production use, create a dedicated bot with specific permissions:

Go to Settings > Bots
Click Add Bot
Provide a name and description
Assign appropriate roles (typically DefaultBotPolicy and Ingestion Bot Policy)
Copy the generated JWT token

Configuring the SDK

Once you have a JWT token, configure the SDK in your Python code.

If your organization uses an external secrets manager for credential storage, initialize it before this first configure() call, not after. SecretsManagerFactory is a singleton: whichever call runs first wins, so initializing it later in Using External Secrets Managers has no effect once configure() has already run here. See that section before continuing if this applies to you.

from metadata.sdk import configure

configure(
    host="http://localhost:8585/api",  # Your OpenMetadata API URL
    jwt_token="your-jwt-token-here"
)

Using Environment Variables

For better security, let configure pick them up from environment variables:

from metadata.sdk import configure

configure()

Set the environment variable before running your script:

export OPENMETADATA_HOST="http://localhost:8585/api"
export OPENMETADATA_JWT_TOKEN="your-jwt-token-here"

python your_script.py

Configuration Parameters

The configure() function accepts the following parameters:

Parameter	Type	Required	Description	Environment Variable
`host`	`str`	No	OpenMetadata API URL (e.g., `http://localhost:8585/api`)	`OPENMETADATA_HOST`
`jwt_token`	`str`	No	JWT authentication token	`OPENMETADATA_JWT_TOKEN`

Using External Secrets Managers

This section explains when and how to configure the SDK to work with an external secrets manager instead of database-stored credentials.

Important Note

If your OpenMetadata instance uses database-stored credentials (the default configuration), you don’t need to follow this guide. The SDK will automatically retrieve and decrypt credentials. This guide is only necessary when your organization uses an external secrets manager for credential storage.

Why This Is Required

The TestRunner API executes data quality tests directly from your Python code (for example, within your ETL pipelines). To connect to your data sources, it needs to:

Retrieve the service connection configuration from OpenMetadata
Decrypt the credentials stored in your secrets manager
Establish a connection to the data source
Execute the test cases

Without proper secrets manager configuration, the SDK cannot decrypt credentials and will fail to connect to your data sources.

General Setup Steps

Contact your OpenMetadata administrator to obtain:
- The secrets manager type (AWS, Azure, GCP, and so on)
- The secrets manager loader configuration
- Required environment variables or configuration files
- Any additional setup (IAM roles, service principals, and so on)
Install required dependencies for your secrets manager provider
Configure environment variables with access credentials
Initialize the SecretsManagerFactory before using TestRunner
Authenticate with an ingestion-bot JWT instead of a personal user token. See Authentication for how to obtain one.
Configure the SDK
Run your tests

Example Using AWS Secrets Manager

Required Dependencies:

pip install "openmetadata-ingestion~=2.0.3.0"

The base ingestion package includes secrets manager client libraries. Install any connector-specific extras required by your test YAML separately, for example openmetadata-ingestion[mysql,snowflake]~=2.0.3.0. Make sure to use the same openmetadata-ingestion version as your OpenMetadata server version. Example Configuration:

import os

from metadata.generated.schema.security.secrets.secretsManagerClientLoader import SecretsManagerClientLoader
from metadata.generated.schema.security.secrets.secretsManagerProvider import SecretsManagerProvider
from metadata.sdk import configure
from metadata.sdk.data_quality import TestRunner
from metadata.utils.secrets.secrets_manager_factory import SecretsManagerFactory

# Set AWS credentials and region
os.environ["AWS_ACCESS_KEY_ID"] = "your-access-key-id"
os.environ["AWS_SECRET_ACCESS_KEY"] = "your-secret-access-key"
os.environ["AWS_DEFAULT_REGION"] = "us-east-1"  # Your AWS region

# Initialize secrets manager (must be done before configure())
SecretsManagerFactory(
    secrets_manager_provider=SecretsManagerProvider.managed_aws,
    secrets_manager_loader=SecretsManagerClientLoader.env,
)

# Configure OpenMetadata SDK
configure(
    host="https://your-openmetadata-instance.com/api",
    jwt_token="your-jwt-token",
)

# Use TestRunner as normal
runner = TestRunner.for_table("MySQL.production.database.my_table")
results = runner.run()

Configuration by Provider

Each provider needs its own dependency, SecretsManagerProvider value, and environment variables, listed below.

AWS Secrets Manager and AWS Systems Manager Parameter Store

OpenMetadata ingestion dependency: the base openmetadata-ingestion package includes the AWS secrets manager client libraries. SecretsManagerProvider: (one of)

SecretsManagerProvider.aws
SecretsManagerProvider.managed_aws
SecretsManagerProvider.aws_ssm
SecretsManagerProvider.managed_aws_ssm

Environment variables:

AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY
AWS_DEFAULT_REGION

Azure Key Vault

OpenMetadata ingestion dependency: the base openmetadata-ingestion package includes the Azure Key Vault client libraries. SecretsManagerProvider: (one of)

SecretsManagerProvider.azure_kv
SecretsManagerProvider.managed_azure_kv

Environment variables:

AZURE_CLIENT_ID
AZURE_CLIENT_SECRET
AZURE_TENANT_ID
AZURE_KEY_VAULT_NAME

Google Cloud Secret Manager

OpenMetadata ingestion dependency: the base openmetadata-ingestion package includes the Google Cloud Secret Manager client libraries. SecretsManagerProvider: SecretsManagerProvider.gcp Environment variables:

GOOGLE_APPLICATION_CREDENTIALS: path to the credentials JSON file
GOOGLE_CLOUD_PROJECT

Troubleshooting

If you hit one of these errors while using an external secrets manager, check the matching cause and solution below.

Error: “Cannot decrypt service connection”

Cause: Secrets manager not initialized or misconfigured Solution: Ensure SecretsManagerFactory is initialized before calling configure() or creating the TestRunner

Error: “SecretsManagerFactory settings don’t take effect”

Cause: SecretsManagerFactory is a singleton. Only the first call in a Python process takes effect. If configure(), TestRunner, or anything else from metadata.sdk runs first, the factory already initializes with defaults. This happens even if the first call comes through an earlier import. Later SecretsManagerFactory(...) calls are silently ignored. Solution: Call SecretsManagerFactory(...) as the first SDK-related statement in your script. Restart the session if you’re in a long-running or interactive environment, such as a notebook, where the SDK might already have been used.

Error: “Access Denied” or “Unauthorized”

Cause: Insufficient permissions to access secrets Solution:

Verify IAM role/service principal has correct permissions
Check credentials are valid and not expired
Ensure correct region/vault name is specified

Error: “Module not found” for secrets manager

Cause: Missing dependencies for your secrets manager Solution: Install the base ingestion package that matches your OpenMetadata server version. Add connector-specific extras required by the service you test.

pip install "openmetadata-ingestion~=2.0.3.0"
pip install "openmetadata-ingestion[mysql,snowflake,s3]~=2.0.3.0"

Tests Fail with Connection Errors

Cause: Credentials not properly decrypted or secrets manager misconfigured Solution:

Verify secrets manager provider matches your OpenMetadata backend configuration
Test credential access independently (for example, using AWS CLI, Azure CLI, or gcloud)
Check network connectivity to secrets manager service
Enable debug logging to see detailed error messages:

import logging
logging.basicConfig(level=logging.DEBUG)

Contact Your Administrator

If you’re unsure about:

Which secrets manager your organization uses
Required environment variables or configuration
Access credentials or IAM roles
Permissions needed

Contact your OpenMetadata administrator for the specific configuration required in your environment.

For More Information

For more information about configuring a secrets manager for your OpenMetadata deployment, see Enable Secrets Manager.
For provider-specific setup steps, see Supported implementations.

Verify Installation

Create a simple test to verify your setup:

from metadata.sdk import configure
from metadata.sdk.data_quality import TestRunner

# Configure SDK
configure(
    host="http://localhost:8585/api",
    jwt_token="your-jwt-token-here"
)

# Test connection by creating a runner
try:
    runner = TestRunner.for_table("your_service.database.schema.table")
    print("✓ SDK configured successfully!")
except Exception as e:
    print(f"✗ Configuration failed: {e}")

Replace "your_service.database.schema.table" with the fully qualified name of an actual table in your OpenMetadata instance.

Your First Data Quality Test

Now that you’re set up, let’s run your first data quality test:

from metadata.sdk import configure
from metadata.sdk.data_quality import TestRunner, TableRowCountToBeBetween

# Configure SDK
configure(
    host="http://localhost:8585/api",
    jwt_token="your-jwt-token-here"
)

# Create a test runner for a specific table
runner = TestRunner.for_table("MySQL.ecommerce.public.customers")

# Add a test to verify row count is within expected range
runner.add_test(
    TableRowCountToBeBetween(min_count=1000, max_count=100000)
)

# Run the tests
results = runner.run()

# Print results
for result in results:
    test_case = result.testCase
    test_result = result.testCaseResult

    print(f"Test: {test_case.name.root}")
    print(f"Status: {test_result.testCaseStatus}")
    print(f"Result: {test_result.result}")

Common Installation Issues

Connection Timeout

If you experience connection timeouts, verify:

OpenMetadata instance is running and accessible
API URL is correct (should end with /api)
Network connectivity between your script and OpenMetadata
Firewall rules allow the connection

Import Errors

If you encounter import errors:

ModuleNotFoundError: No module named 'metadata'

Verify the package is installed correctly:

pip list | grep openmetadata

If not listed, reinstall:

pip install --upgrade "openmetadata-ingestion~=2.0.3.0"

Next Steps

Now that you have the SDK installed and configured:

Learn how to run table-level tests using the TestRunner API
Explore DataFrame validation for ETL pipelines
Review the complete test definitions reference

Data Quality and Observability

Getting Started with Data Quality as Code

Getting Started with Data Quality as Code

Prerequisites

Installation

Basic Installation

Installation with Database Connectors

Installation with DataFrame Support

Installation with Multiple Features

Authentication

Getting a JWT Token

Option 1: Using an Existing Bot Token

Option 2: Creating a Custom Bot

Configuring the SDK

Using Environment Variables

Configuration Parameters

Using External Secrets Managers

Important Note

Why This Is Required

General Setup Steps

Example Using AWS Secrets Manager

Configuration by Provider

AWS Secrets Manager and AWS Systems Manager Parameter Store

Azure Key Vault

Google Cloud Secret Manager

Troubleshooting

Error: “Cannot decrypt service connection”

Error: “SecretsManagerFactory settings don’t take effect”

Error: “Access Denied” or “Unauthorized”

Error: “Module not found” for secrets manager

Tests Fail with Connection Errors

Contact Your Administrator

For More Information

Verify Installation

Your First Data Quality Test

Common Installation Issues

Connection Timeout

Import Errors

Next Steps

Additional Resources

​Getting Started with Data Quality as Code

​Prerequisites

​Installation

​Basic Installation

​Installation with Database Connectors

​Installation with DataFrame Support

​Installation with Multiple Features

​Authentication

​Getting a JWT Token

​Option 1: Using an Existing Bot Token

​Option 2: Creating a Custom Bot

​Configuring the SDK

​Using Environment Variables

​Configuration Parameters

​Using External Secrets Managers

​Important Note

​Why This Is Required

​General Setup Steps

​Example Using AWS Secrets Manager

​Configuration by Provider

​AWS Secrets Manager and AWS Systems Manager Parameter Store

​Azure Key Vault

​Google Cloud Secret Manager

​Troubleshooting

​Error: “Cannot decrypt service connection”

​Error: “SecretsManagerFactory settings don’t take effect”

​Error: “Access Denied” or “Unauthorized”

​Error: “Module not found” for secrets manager

​Tests Fail with Connection Errors

​Contact Your Administrator

​For More Information

​Verify Installation

​Your First Data Quality Test

​Common Installation Issues

​Connection Timeout

​Import Errors

​Next Steps

​Additional Resources

Getting Started with Data Quality as Code

Prerequisites

Installation

Basic Installation

Installation with Database Connectors

Installation with DataFrame Support

Installation with Multiple Features

Authentication

Getting a JWT Token

Option 1: Using an Existing Bot Token

Option 2: Creating a Custom Bot

Configuring the SDK

Using Environment Variables

Configuration Parameters

Using External Secrets Managers

Important Note

Why This Is Required

General Setup Steps

Example Using AWS Secrets Manager

Configuration by Provider

AWS Secrets Manager and AWS Systems Manager Parameter Store

Azure Key Vault

Google Cloud Secret Manager

Troubleshooting

Error: “Cannot decrypt service connection”

Error: “SecretsManagerFactory settings don’t take effect”

Error: “Access Denied” or “Unauthorized”

Error: “Module not found” for secrets manager

Tests Fail with Connection Errors

Contact Your Administrator

For More Information

Verify Installation

Your First Data Quality Test

Common Installation Issues

Connection Timeout

Import Errors

Next Steps

Additional Resources