refactor extraction block in an util function

Author: themisvaltinos

sqlmesh/core/dialect.py

@@ -1408,6 +1408,41 @@ def extract_func_call(
     return func.lower(), kwargs
 
 
+def extract_function_calls(func_calls: t.Any, allow_tuples: bool = False) -> t.Any:
+    """Used for extracting function calls for signals or audits."""
+
+    if isinstance(func_calls, (exp.Tuple, exp.Array)):
+        return [extract_func_call(i, allow_tuples=allow_tuples) for i in func_calls.expressions]
+    if isinstance(func_calls, exp.Paren):
+        return [extract_func_call(func_calls.this, allow_tuples=allow_tuples)]
+    if isinstance(func_calls, exp.Expression):
+        return [extract_func_call(func_calls, allow_tuples=allow_tuples)]
+    if isinstance(func_calls, list):
+        function_calls = []
+        for entry in func_calls:
+            if isinstance(entry, dict):
+                args = entry
+                name = "" if allow_tuples else entry.pop("name")
+            elif isinstance(entry, (tuple, list)):
+                name, args = entry
+            else:
+                raise ConfigError(f"Audit must be a dictionary or named tuple. Got {entry}.")
+
+            function_calls.append(
+                (
+                    name.lower(),
+                    {
+                        key: parse_one(value) if isinstance(value, str) else value
+                        for key, value in args.items()
+                    },
+                )
+            )
+
+        return function_calls
+
+    return func_calls or []
+
+
 def is_meta_expression(v: t.Any) -> bool:
     return isinstance(v, (Audit, Metric, Model))
 

sqlmesh/core/model/definition.py

@@ -2539,21 +2539,8 @@ def _create_model(
             used_variables.update(extract_macro_references_and_variables(jinja_macro.definition)[1])
 
     # Merge model-specific audits with default audits
-    model_audits = kwargs.pop("audits", [])
-    default_audits = defaults.pop("audits", [])
-
-    if isinstance(model_audits, (exp.Tuple, exp.Array)):
-        model_audits_list = [d.extract_func_call(i) for i in model_audits.expressions]
-    elif isinstance(model_audits, exp.Paren):
-        model_audits_list = [d.extract_func_call(model_audits.this)]
-    elif isinstance(model_audits, exp.Expression):
-        model_audits_list = [d.extract_func_call(model_audits)]
-    elif isinstance(model_audits, list):
-        model_audits_list = model_audits
-    else:
-        model_audits_list = []
-    merged_audits = default_audits + model_audits_list
-    kwargs["audits"] = merged_audits
+    if default_audits := defaults.pop("audits", None):
+        kwargs["audits"] = default_audits + d.extract_function_calls(kwargs.pop("audits", []))
 
     model = klass(
         name=name,

sqlmesh/core/model/meta.py

@@ -11,7 +11,7 @@
 
 from sqlmesh.core import dialect as d
 from sqlmesh.core.config.linter import LinterConfig
-from sqlmesh.core.dialect import normalize_model_name, extract_func_call
+from sqlmesh.core.dialect import normalize_model_name
 from sqlmesh.core.model.common import (
     bool_validator,
     default_catalog_validator,
@@ -94,37 +94,7 @@ class ModelMeta(_Node):
     def _func_call_validator(cls, v: t.Any, field: t.Any) -> t.Any:
         is_signal = getattr(field, "name" if hasattr(field, "name") else "field_name") == "signals"
 
-        if isinstance(v, (exp.Tuple, exp.Array)):
-            return [extract_func_call(i, allow_tuples=is_signal) for i in v.expressions]
-        if isinstance(v, exp.Paren):
-            return [extract_func_call(v.this, allow_tuples=is_signal)]
-        if isinstance(v, exp.Expression):
-            return [extract_func_call(v, allow_tuples=is_signal)]
-        if isinstance(v, list):
-            audits = []
-
-            for entry in v:
-                if isinstance(entry, dict):
-                    args = entry
-                    name = "" if is_signal else entry.pop("name")
-                elif isinstance(entry, (tuple, list)):
-                    name, args = entry
-                else:
-                    raise ConfigError(f"Audit must be a dictionary or named tuple. Got {entry}.")
-
-                audits.append(
-                    (
-                        name.lower(),
-                        {
-                            key: d.parse_one(value) if isinstance(value, str) else value
-                            for key, value in args.items()
-                        },
-                    )
-                )
-
-            return audits
-
-        return v or []
+        return d.extract_function_calls(v, allow_tuples=is_signal)
 
     @field_validator("tags", mode="before")
     def _value_or_tuple_validator(cls, v: t.Any, info: ValidationInfo) -> t.Any: