Feat: Add support for defaults in user audits (#2901)

Author: alexischicoinedatadeveloper

docs/concepts/audits.md

@@ -75,6 +75,19 @@ Notice how `column` and `threshold` parameters have been set. These values will
 
 Note that the same audit can be applied more than once to the a model using different sets of parameters.
 
+Generic audits can define default values as follows:
+```sql linenums="1"
+AUDIT (
+  name does_not_exceed_threshold,
+  defaults (
+    threshold = 10,
+    column = id
+  )
+);
+SELECT * FROM @this_model
+WHERE @column >= @threshold;
+```
+
 ### Naming
 We recommended avoiding SQL keywords when naming audit parameters. Quote any audit argument that is also a SQL keyword.
 

sqlmesh/core/audit/definition.py

@@ -34,6 +34,7 @@
 from sqlmesh.utils.pydantic import (
     PydanticModel,
     field_validator,
+    get_dialect,
     model_validator,
     model_validator_v1_args,
 )
@@ -148,11 +149,19 @@ def audit_string_validator(cls: t.Type, v: t.Any) -> t.Optional[str]:
 
 
 @field_validator("defaults", mode="before", check_fields=False)
-def audit_map_validator(cls: t.Type, v: t.Any) -> t.Dict[str, t.Any]:
+def audit_map_validator(cls: t.Type, v: t.Any, values: t.Any) -> t.Dict[str, t.Any]:
+    if isinstance(v, exp.Paren):
+        return dict([_maybe_parse_arg_pair(v.unnest())])
     if isinstance(v, (exp.Tuple, exp.Array)):
         return dict(map(_maybe_parse_arg_pair, v.expressions))
     elif isinstance(v, dict):
-        return v
+        dialect = get_dialect(values)
+        return {
+            key: value
+            if isinstance(value, exp.Expression)
+            else d.parse_one(str(value), dialect=dialect)
+            for key, value in v.items()
+        }
     else:
         raise_config_error(
             "Defaults must be a tuple of exp.EQ or a dict", error_type=AuditConfigError

tests/core/test_audit.py

@@ -303,6 +303,49 @@ def test_load_multiple(assert_exp_eq):
     )
 
 
+def test_load_with_dictionary_defaults():
+    expressions = parse(
+        """
+        AUDIT (
+            name my_audit,
+            dialect spark,
+            defaults (
+                field1 = some_column,
+                field2 = 3
+            ),
+        );
+
+        SELECT 1
+    """
+    )
+
+    audit = load_audit(expressions, dialect="spark")
+    assert audit.defaults.keys() == {"field1", "field2"}
+    for value in audit.defaults.values():
+        assert isinstance(value, exp.Expression)
+
+
+def test_load_with_single_defaults():
+    # testing it also works with a single default with no trailing comma
+    expressions = parse(
+        """
+        AUDIT (
+            name my_audit,
+            defaults (
+                field1 = some_column
+            ),
+        );
+
+        SELECT 1
+    """
+    )
+
+    audit = load_audit(expressions, dialect="duckdb")
+    assert audit.defaults.keys() == {"field1"}
+    for value in audit.defaults.values():
+        assert isinstance(value, exp.Expression)
+
+
 def test_no_audit_statement():
     expressions = parse(
         """