Add option to store audit failures in database

[fredguth]

Store SQLMesh Audit Failures for Later Consumption

Summary

Add an option to sqlmesh audit to store failed audit results in a table for later analysis and reporting.

Background

Similar features exist in other data tools:

• dbt: store_failures
• Dataform: Assertions

Use Case

Storing failed audit results enables:

• Historical tracking of audit performance
• Integration with data catalogs and quality dashboards
• Detailed analysis of failure patterns

Example: Generating a data catalog with a "Quality" tab showing all audits and their results, similar to Great Expectations Cloud.

Proposed Solution

1. Create a new table for failed audits

SELECT * FROM audits.failed

Example output:

project	model	audit_id	audit_description	query	failed	audited_at
Project A	Model X	audit_1	Initial audit	SELECT * ...	true	2024-08-28 07:56:36.366-03
Project B	Model Y	audit_2	...	...	false	2024-08-28 07:56:36.366-03
Project C	Model Y	audit_3	...	...	false	2024-08-28 07:56:36.366-03

2. Store detailed failure information

For each failed audit, create a separate table with failure details:

SELECT * FROM audits.audit_1

Alternative: JSON Export

Consider exporting failed values as a JSON object for flexibility.

Implementation Details

• Add a new command-line option to sqlmesh audit (e.g., --store-failures)
• Implement logic to create and populate the audits.failed table
• Create individual tables for each failed audit
• (Optional) Implement JSON export functionality

Questions

• How long should failed audit data be retained?
• Is there a way to store any kind of audit failure in the same assertion_table independent of the model and the audit query?

[fredguth]

Are there any plans to implement this feature? In the meantime, is there a recommended temporary solution?

[tobymao]

no there are no plans for this since this is already in in tobiko cloud

[Harmuth94]

Are there any plans to implement this feature? In the meantime, is there a recommended temporary solution?

Going by the implementation of the audit command, it calls the audit function in context.py

What you basically want is to use the python interface to SQLMesh and manually run the audits as defined in the context.audit() function.

from sqlmesh.core.context import Context
context = Context(paths="example")
start, end = "yesterday", "now"

snapshots = context.snapshots.values()

for snapshot in snapshots:
    for audit_result in context.snapshot_evaluator.audit(
        snapshot=snapshot,
        start=start,
        end=end,
        snapshots=context.snapshots,
        raise_exception=False,
    ):
        if audit_result.count:
            if audit_result.query:
                query_with_failed_results = audit_result.query
                # use the query to save to table, insert, etc.

[fredguth]

Going by the implementation of the audit command, it calls the audit function in context.py

What you basically want is to use the python interface to SQLMesh and manually run the audits as defined in the context.audit() function.

@Harmuth94 Did you have the same problem? I understood you took the approach you laid down and created your own audit command, right?

[Harmuth94]

Well actually we changed the audits to models with a validated column (to persist) and then a simple audit on top to select id from table where not validated

[christophmeier90]

@tobymao Can you explain how this is implemented in Tobiko Cloud?
Does it capture logs and sink them somewhere?
I understand persistence of audit results won't be implemented for core SQLMesh, but it's still very valuable for most teams. We should at least be able to do it ourselves.
The above suggestion by @Harmuth94 involves using the Python client to run all audits manually, but that duplicates cost (model runs already execute audits) and risks drift (results won't represent the true results of tests that ran)

[christophmeier90]

@fredguth @Harmuth94
In case anyone is interested, this problem can be solved using the class sqlmesh.core.notification_target.GenericNotificationTarget. It lets you intercept a certain family of events and do whatever you want with them.

This method builds upon this article to handle notifications, but uses GenericNotificationTarget, which can handle things beyond Slack and SMTP :
https://sqlmesh.readthedocs.io/en/stable/guides/notifications/

Below is an example that writes failing audits to a csv file in a specified path.

# custom_notifications/audit_failure.py
from __future__ import annotations

import csv
import datetime as dt
import os
import typing as t

from pydantic import Field
from sqlmesh.core.notification_target import GenericNotificationTarget, NotificationStatus, NotificationEvent
from sqlmesh.utils.errors import AuditError


class AuditFailureCsvTarget(GenericNotificationTarget, frozen=True):
    # Must stay "generic" to pass Config's union discriminator.
    type_: t.Literal["generic"] = Field(alias="type", default="generic")

    path: str
    notify_on: t.FrozenSet[NotificationEvent] = frozenset({NotificationEvent.AUDIT_FAILURE})

    def send(
        self,
        notification_status: NotificationStatus,
        msg: str,
        audit_error: t.Optional[AuditError] = None,
        exc: t.Optional[str] = None,
        **kwargs: t.Any,
    ) -> None:
        if not audit_error:
            return

        row = {
            "event_timestamp": dt.datetime.now(dt.timezone.utc).isoformat(),
            "model": getattr(audit_error, "model_name", "") or "",
            "audit": getattr(audit_error, "audit_name", "") or "",
            "count": getattr(audit_error, "count", None),
            "blocking": getattr(audit_error, "blocking", None),
            "message": str(audit_error),
            "query": audit_error.sql(pretty=False) if hasattr(audit_error, "sql") else "",
        }

        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
        write_header = not os.path.exists(self.path) or os.path.getsize(self.path) == 0

        try:
            with open(self.path, "a", newline="", encoding="utf-8") as f:
                w = csv.DictWriter(f, fieldnames=list(row.keys()))
                if write_header:
                    w.writeheader()
                w.writerow(row)
        except Exception:
            return

Then, in your config.py, you'd wire this in like this:

from custom_notifications.audit_failure import AuditFailureCsvTarget

config = Config(
    [...]

    notification_targets=[
        AuditFailureCsvTarget(
            notify_on=["audit_failure"], 
            path="observability/audit_failures.csv",
        )
    ],
)

Of course, you can modify this behavior to do anything you want. You could send those events to Kafka, or write them to the state database yourself, if you create a table for them. How you implement this is entirely up to you.