sqlmesh.core.audit

 1import inspect
 2import typing as t
 3from types import ModuleType
 4
 5from sqlmesh.core.audit import builtin
 6from sqlmesh.core.audit.definition import (
 7    Audit as Audit,
 8    AuditResult as AuditResult,
 9    ModelAudit as ModelAudit,
10    StandaloneAudit as StandaloneAudit,
11    load_audit as load_audit,
12    load_multiple_audits as load_multiple_audits,
13)
14
15
16def create_non_blocking_copy(audit: Audit) -> Audit:
17    return audit.copy(update={"name": f"{audit.name}_non_blocking", "blocking": False})
18
19
20def _discover_audits(modules: t.Iterable[ModuleType]) -> t.Dict[str, Audit]:
21    return {
22        audit.name: audit
23        for module in modules
24        for _, model_audit in inspect.getmembers(module, lambda v: isinstance(v, ModelAudit))
25        for audit in (model_audit, create_non_blocking_copy(model_audit))
26    }
27
28
29BUILT_IN_AUDITS = _discover_audits([builtin])