Implementation:TobikoData Sqlmesh Context Audit

Knowledge Sources	SQLMesh SQLMesh Docs
Domains	Data_Engineering, Data_Quality, Model_Development
Last Updated	2026-02-07 00:00 GMT

Overview

Concrete audit execution engine for running data quality checks provided by SQLMesh.

Description

The Context.audit method executes data quality checks (audits) against model outputs within a specified time interval. It resolves which models to audit (all models if none specified), collects all audits associated with those models, executes each audit query against the database, evaluates pass/fail conditions, and reports results with detailed status information for passed, failed, skipped, and errored audits.

Audits are queries that validate data quality assumptions—checking for nulls, validating row counts, ensuring referential integrity, or detecting anomalies. This method orchestrates the execution of all audits, handles errors gracefully, tracks which audits should be skipped based on data availability, and aggregates results into a comprehensive quality report. It returns a boolean indicating overall success (all audits passed) or failure (one or more audits failed).

Usage

Use this method during development to validate that model changes maintain data quality, in CI/CD pipelines to prevent deploying models that violate quality constraints, and on production schedules to continuously monitor data quality. Run audits after backfilling data to ensure correctness before promoting changes to production.

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/core/context.py (Lines 2271-2313)

Signature

def audit(
    self,
    start: TimeLike,
    end: TimeLike,
    *,
    models: t.Optional[t.Iterator[str]] = None,
    execution_time: t.Optional[TimeLike] = None,
) -> bool

Import

from sqlmesh import Context

context = Context()

I/O Contract

Inputs

Name	Type	Required	Description
start	TimeLike	Yes	Start of the time interval to audit
end	TimeLike	Yes	End of the time interval to audit
models	Iterator[str]	No	Specific model names to audit (default: all models in environment)
execution_time	TimeLike	No	Reference timestamp for execution (default: now)

Outputs

Name	Type	Description
success	bool	True if all audits passed, False if any audit failed or errored

Usage Examples

Basic Audit Execution

from sqlmesh import Context

context = Context()

# Run all audits for a specific date range
success = context.audit(
    start="2024-01-01",
    end="2024-01-31"
)

if success:
    print("All audits passed!")
else:
    print("Some audits failed - check logs for details")
    exit(1)

Auditing Specific Models

# Run audits only for revenue-related models
success = context.audit(
    start="2024-01-01",
    end="2024-01-07",
    models=["my_schema.revenue_model", "my_schema.customer_revenue"]
)

print(f"Revenue model audits: {'PASSED' if success else 'FAILED'}")

Development Workflow with Audits

# After making changes, validate data quality

# 1. Create dev environment
plan = context.plan(
    environment="dev",
    create_from="prod",
    start="2024-01-01",
    end="2024-01-07",
    auto_apply=True
)

# 2. Run audits in dev environment
success = context.audit(
    start="2024-01-01",
    end="2024-01-07"
)

if success:
    print("Dev environment data quality validated!")
    # Safe to promote to production
else:
    print("Data quality issues detected - fix before promoting")

CI/CD Pipeline Integration

import sys
from sqlmesh import Context

def run_audit_checks():
    """Run audits as part of CI/CD pipeline"""
    context = Context()

    # Create temporary test environment
    context.plan(
        environment="ci_test",
        create_from="prod",
        start="2024-01-01",
        end="2024-01-01",  # Single day for fast CI
        auto_apply=True,
        no_prompts=True
    )

    # Run audits
    success = context.audit(
        start="2024-01-01",
        end="2024-01-01",
        execution_time="2024-01-02"
    )

    if not success:
        print("ERROR: Audits failed in CI")
        sys.exit(1)

    print("CI audits passed successfully")
    return 0

if __name__ == "__main__":
    exit(run_audit_checks())

Scheduled Quality Monitoring

from datetime import datetime, timedelta
from sqlmesh import Context

def daily_audit_check():
    """Run daily audit checks on production data"""
    context = Context()

    # Audit yesterday's data
    yesterday = datetime.now() - timedelta(days=1)
    start = yesterday.strftime("%Y-%m-%d")
    end = start

    success = context.audit(
        start=start,
        end=end,
        execution_time=datetime.now()
    )

    if not success:
        # Send alert notification
        send_alert(f"Data quality audit failed for {start}")
        return False

    print(f"Daily audit for {start} completed successfully")
    return True

# Schedule this to run daily
daily_audit_check()

Auditing After Model Changes

# After refactoring a model, validate it maintains quality

# 1. Test model logic
results = context.evaluate(
    "my_schema.refactored_model",
    start="2024-01-01",
    end="2024-01-01",
    execution_time="2024-01-02"
)
print(f"Model returned {len(results)} rows")

# 2. Run audits to validate quality constraints
success = context.audit(
    start="2024-01-01",
    end="2024-01-01",
    models=["my_schema.refactored_model"]
)

if success:
    print("Refactoring maintained data quality!")
else:
    print("Refactoring broke data quality - reverting changes")

Audit with Custom Execution Time

# Run audits as of a specific point in time
# Useful for validating historical data

success = context.audit(
    start="2024-01-01",
    end="2024-01-31",
    execution_time="2024-02-01 12:00:00"  # Specific execution timestamp
)

print(f"Historical audit status: {'PASSED' if success else 'FAILED'}")

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Audit_Execution