Metadata

Overview

The Paradime Python SDK Metadata module helps you query and analyze dbt™ metadata from Bolt runs.

Use it to monitor model health, analyze dbt test results, track source freshness, and explore upstream/downstream lineage. It uses a local DuckDB database and returns results as high-performance polars DataFrames.

• Query dbt artifacts like manifest.json, run_results.json, and source freshness results

• Build reliability checks around failing models, failing tests, and stale sources

• Run ad-hoc SQL with query_sql() against the DuckDB metadata schema

The Metadata SDK is available in Python SDK version 4.18.0+ and requires a Paradime account with access to dbt metadata.

Quick Start

Basic Setup

from paradime.client.paradime_client import Paradime

client = Paradime(
    api_endpoint="your-api-endpoint",
    api_key="your-api-key",
    api_secret="your-api-secret"
)

metadata = client.metadata

Simple Health Check

models = metadata.get_model_health("daily_production_run")

successful = len([m for m in models if m.status.value == "success"])
print(f"✅ {successful}/{len(models)} models successful")

Model Health Monitoring

Get Model Health Status

Monitor the health of all models in a dbt run, including execution status, test results, and performance metrics.

models = metadata.get_model_health("schedule_name")

for model in models:
    print(f"{model.name}")
    print(f"  Status: {model.status.value}")
    print(f"  Health: {model.health_status.value}")
    print(f"  Tests: {model.total_tests} total, {model.failed_tests} failed")
    print(f"  Execution: {model.execution_time}s")

Filter by Health Status

Failing models

failing_models = metadata.get_failing_models("schedule_name")

for model in failing_models:
    print(f"{model.name} — {model.failed_tests} failed tests")
    if model.error_message:
        print(f"  Error: {model.error_message}")

Slowest models

Test Results

# All test results
tests = metadata.get_test_results("schedule_name")

# Failed tests only
failed_tests = metadata.get_test_results("schedule_name", failed_only=True)

for test in failed_tests:
    print(f"❌ {test.test_name}")
    print(f"   Status: {test.status.value}")
    print(f"   Models: {test.depends_on_nodes}")

Source Freshness

sources = metadata.get_source_freshness("schedule_name")

for source in sources:
    print(f"{source.source_name}.{source.table_name}")
    print(f"  Status: {source.freshness_status.value}")
    print(f"  Last loaded: {source.max_loaded_at}")
    print(f"  Hours since load: {source.hours_since_load}")

Sources are loaded from a dedicated source freshness schedule (e.g. one running dbt source freshness), not from your main dbt run schedule.

Custom SQL Queries

The metadata SDK exposes a DuckDB database you can query directly using SQL. Results are returned as polars DataFrames.

sql = """
    SELECT
        schema_name,
        COUNT(*) as model_count,
        AVG(execution_time) as avg_execution_time
    FROM dbt_run_results
    WHERE schedule_name = ? AND resource_type = 'model'
    GROUP BY schema_name
    ORDER BY model_count DESC
"""

results = metadata.query_sql(sql, "schedule_name", ["schedule_name"])

for row in results.iter_rows(named=True):
    print(f"{row['schema_name']}: {row['model_count']} models")

query_sql returns a polars.DataFrame. Use .iter_rows(named=True) to iterate with dict-style row access.

Dependency Analysis

Upstream Dependencies

dependencies = metadata.get_upstream_health(
    model_name="my_important_model",
    schedule_name="schedule_name",
    max_depth=5
)

for dep in dependencies:
    print(f"Level {dep.level}: {dep.name}")
    print(f"  Health: {dep.health_status.value}")
    print(f"  Status: {dep.status.value}")

Downstream Impact

impact = metadata.get_downstream_impact(
    model_name="failed_model",
    schedule_name="schedule_name",
    max_depth=5
)

print(f"Impact of {impact.failed_model} failure:")
print(f"  Critical: {len(impact.critical_models)}")
print(f"  Warning: {len(impact.warning_models)}")
print(f"  Potentially affected: {len(impact.potentially_affected)}")

Health Dashboard

dashboard = metadata.get_health_dashboard("schedule_name")

print(f"Models — Total: {dashboard.total_models}")
print(f"  Healthy: {dashboard.healthy_models}")
print(f"  Warning: {dashboard.warning_models}")
print(f"  Critical: {dashboard.critical_models}")
print(f"Avg execution: {dashboard.avg_execution_time:.1f}s")
print(f"Test success rate: {dashboard.test_success_rate:.1f}%")

Advanced Features

Performance Summary

perf = metadata.get_performance_summary("schedule_name", days=7)

print(f"Avg execution: {perf.average_execution_time:.1f}s")
print(f"Success rate: {perf.success_rate:.1f}%")

for model in perf.slowest_models[:5]:
    print(f"  {model['name']}: {model['execution_time']:.1f}s")

Streaming Large Datasets

For large metadata datasets, stream results in batches to avoid memory pressure.

for batch in metadata.get_model_health_stream("schedule_name", batch_size=100):
    for model in batch:
        if model.health_status.value == "Critical":
            print(f"Critical: {model.name}")

Cache Management

metadata.clear_cache()                    # clear all
metadata.clear_cache("schedule_name")     # clear one schedule
metadata.refresh_metadata("schedule_name") # force refresh

Configuration

Custom Database Connection

By default the SDK uses an in-memory DuckDB instance. Use a file path for persistent storage across sessions.

from paradime.apis.metadata.client import MetadataClient

metadata_client = MetadataClient(
    bolt_client=client.bolt,
    db_connection="./metadata.duckdb",  # persistent
    cache_ttl_seconds=600               # 10 min cache
)

Complete Example

from paradime.client.paradime_client import Paradime

def monitor_dbt_health(schedule_name: str) -> None:
    client = Paradime(
        api_endpoint="your-api-endpoint",
        api_key="your-api-key",
        api_secret="your-api-secret"
    )
    metadata = client.metadata

    dashboard = metadata.get_health_dashboard(schedule_name)
    print(f"Models: {dashboard.healthy_models}✅  {dashboard.warning_models}⚠️  {dashboard.critical_models}❌")

    if dashboard.critical_models > 0:
        for model in metadata.get_failing_models(schedule_name)[:5]:
            print(f"  {model.name}: {model.failed_tests} failed tests")

    for model in metadata.get_slowest_models(schedule_name, limit=3):
        print(f"  {model.name}: {model.execution_time:.1f}s")

    stale = [s for s in metadata.get_source_freshness(schedule_name)
             if s.freshness_status.value != "pass"]
    for source in stale[:3]:
        print(f"  {source.source_name}.{source.table_name}: {source.hours_since_load:.1f}h stale")

    client.metadata.close()

monitor_dbt_health("daily_production_run")

Database Schema

The Metadata SDK loads dbt artifacts into a local DuckDB database. You can query any of these tables directly via query_sql.

Core Tables

dbt_run_results

Main table containing execution results for all dbt resources (models, tests, seeds, snapshots).

Column	Type	Description
unique_id	VARCHAR	Unique identifier for the dbt resource
name	VARCHAR	Resource name
resource_type	VARCHAR	model, test, seed, snapshot
status	VARCHAR	success, error, fail, skipped
execution_time	DOUBLE	Execution time in seconds
executed_at	TIMESTAMP	When the resource was executed
schedule_name	VARCHAR	Schedule identifier
depends_on	VARCHAR[]	Array of upstream dependency IDs
error_message	TEXT	Error details if execution failed
schema_name	VARCHAR	Database schema name
database_name	VARCHAR	Database name
materialized_type	VARCHAR	Model materialization type
description	TEXT	Resource description from dbt
tags	VARCHAR[]	Array of dbt tags
meta	JSON	dbt meta configuration
owner	VARCHAR	Resource owner
compiled_sql	TEXT	Compiled SQL
raw_sql	TEXT	Raw SQL from dbt files
columns	JSON	Column information
children_l1	VARCHAR[]	Direct child dependencies
parents_models	VARCHAR[]	Parent model dependencies
parents_sources	VARCHAR[]	Parent source dependencies

dbt_source_freshness_results

Column	Type	Description
unique_id	VARCHAR	Unique source identifier
source_name	VARCHAR	dbt source name
table_name	VARCHAR	Source table name
schedule_name	VARCHAR	Schedule identifier
freshness_status	VARCHAR	pass, warn, error
max_loaded_at	TIMESTAMP	Last data load timestamp
snapshotted_at	TIMESTAMP	When freshness was checked
hours_since_load	DOUBLE	Hours since last data load
error_after_hours	INTEGER	Error threshold in hours
warn_after_hours	INTEGER	Warning threshold in hours
database	VARCHAR	Source database
schema_name	VARCHAR	Source schema
description	TEXT	Source description
meta	JSON	Source metadata
tags	VARCHAR[]	Source tags

model_metadata

Extended model metadata and lineage information from manifest.json.

Column	Type	Description
unique_id	VARCHAR	Model unique identifier
name	VARCHAR	Model name
resource_type	VARCHAR	Always model
depends_on	VARCHAR[]	Direct dependencies
config	JSON	dbt model configuration
tags	VARCHAR[]	Model tags
meta	JSON	Model metadata
description	TEXT	Model description
columns	JSON	Model column definitions
parents	VARCHAR[]	All parent resources
children	VARCHAR[]	All child resources
original_file_path	VARCHAR	Path to dbt model file

Specialized Tables

Table	Description
dbt_test_data	Detailed test execution results with column-level info
dbt_seed_data	Seed file load results
dbt_snapshot_data	Snapshot execution results
dbt_exposure_data	dbt exposure metadata

Optimized Views

models_with_tests

Pre-calculated model health with test counts.

SELECT unique_id, name, status, execution_time,
       total_tests, failed_tests,
       health_status  -- 'Healthy', 'Warning', 'Critical'
FROM models_with_tests
WHERE schedule_name = 'your_schedule'

latest_model_results

dashboard_metrics

Example Queries

Failed tests

SELECT name, failed_tests, total_tests,
       (failed_tests * 100.0 / total_tests) as failure_rate
FROM models_with_tests
WHERE schedule_name = ?
  AND total_tests > 0
ORDER BY failed_tests DESC
LIMIT 10

Performance by schema

Source freshness alerts

Downstream lineage

API Reference

MetadataClient

Method	Description	Returns
get_model_health(schedule_name)	Health status of all models	List[ModelHealth]
get_failing_models(schedule_name)	Models with failed tests	List[ModelHealth]
get_slowest_models(schedule_name, limit=10)	Slowest running models	List[ModelHealth]
get_test_results(schedule_name, failed_only=False)	Test results	List[TestResult]
get_source_freshness(schedule_name)	Source freshness status	List[SourceFreshness]
get_health_dashboard(schedule_name)	Aggregated health metrics	HealthDashboard
get_performance_summary(schedule_name, days=7)	Performance metrics	PerformanceMetrics
get_upstream_health(model_name, schedule_name, max_depth=10)	Upstream dependencies	List[ModelDependency]
get_downstream_impact(model_name, schedule_name, max_depth=10)	Downstream impact	DependencyImpact
get_model_health_stream(schedule_name, batch_size=100)	Stream models in batches	Iterator[List[ModelHealth]]
query_sql(sql, schedule_name, parameters=None)	Custom SQL query	polars.DataFrame
refresh_metadata(schedule_name)	Force refresh from Bolt	None
clear_cache(schedule_name=None)	Clear in-memory cache	None
close()	Close the DuckDB connection	None

Data Models

ModelHealth

class ModelHealth:
    unique_id: str
    name: str
    resource_type: ResourceType
    status: RunStatus            # success, error, fail, skipped
    execution_time: Optional[float]
    health_status: HealthStatus  # Healthy, Warning, Critical
    total_tests: int
    failed_tests: int
    schema_name: Optional[str]
    database_name: Optional[str]
    error_message: Optional[str]
    description: str
    tags: List[str]
    depends_on: List[str]

HealthDashboard

SourceFreshness

Troubleshooting

Common Issues

Schedule not found

schedules = client.bolt.list_schedules()
print([s.name for s in schedules])

No sources returned

Performance issues

Stale data

Error Handling

try:
    models = metadata.get_model_health("schedule_name")
except ValueError as e:
    print(f"Schedule not found or no artifacts: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")
finally:
    metadata.close()

For advanced use cases, use query_sql to run custom SQL directly against the DuckDB metadata database and return results as a polars DataFrame.