Only validate Params when DAG is triggered (#20802)

airflow/exceptions.py

@@ -167,6 +167,10 @@ class SerializationError(AirflowException):
     """A problem occurred when trying to serialize a DAG."""
 
 
+class ParamValidationError(AirflowException):
+    """Raise when DAG params is invalid"""
+
+
 class TaskNotFound(AirflowNotFoundException):
     """Raise when a Task is not available in the system."""
 

airflow/models/dagbag.py

@@ -40,6 +40,7 @@
     AirflowDagCycleException,
     AirflowDagDuplicatedIdException,
     AirflowTimetableInvalid,
+    ParamValidationError,
 )
 from airflow.stats import Stats
 from airflow.utils import timezone
@@ -398,6 +399,8 @@ def _process_modules(self, filepath, mods, file_last_changed_on_disk):
             dag.fileloc = mod.__file__
             try:
                 dag.timetable.validate()
+                # validate dag params
+                dag.params.validate()
                 self.bag_dag(dag=dag, root_dag=dag)
                 found_dags.append(dag)
                 found_dags += dag.subdags
@@ -409,6 +412,7 @@ def _process_modules(self, filepath, mods, file_last_changed_on_disk):
                 AirflowDagCycleException,
                 AirflowDagDuplicatedIdException,
                 AirflowClusterPolicyViolation,
+                ParamValidationError,
             ) as exception:
                 self.log.exception("Failed to bag_dag: %s", dag.fileloc)
                 self.import_errors[dag.fileloc] = str(exception)

airflow/models/param.py

@@ -23,7 +23,7 @@
 from jsonschema import FormatChecker
 from jsonschema.exceptions import ValidationError
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, ParamValidationError
 from airflow.utils.context import Context
 from airflow.utils.types import NOTSET, ArgNotSet
 
@@ -49,31 +49,6 @@ def __init__(self, default: Any = NOTSET, description: Optional[str] = None, **k
         self.description = description
         self.schema = kwargs.pop('schema') if 'schema' in kwargs else kwargs
 
-        if self.has_value:
-            self._validate(self.value, self.schema)
-
-    @staticmethod
-    def _validate(value, schema):
-        """
-        1. Check that value is json-serializable; if not, warn.  In future release we will require
-        the value to be json-serializable.
-        2. Validate ``value`` against ``schema``
-        """
-        try:
-            json.dumps(value)
-        except Exception:
-            warnings.warn(
-                "The use of non-json-serializable params is deprecated and will be removed in "
-                "a future release",
-                DeprecationWarning,
-            )
-        # If we have a value, validate it once. May raise ValueError.
-        if not isinstance(value, ArgNotSet):
-            try:
-                jsonschema.validate(value, schema, format_checker=FormatChecker())
-            except ValidationError as err:
-                raise ValueError(err)
-
     def __copy__(self) -> "Param":
         return Param(self.value, self.description, schema=self.schema)
 
@@ -82,24 +57,34 @@ def resolve(self, value: Any = NOTSET, suppress_exception: bool = False) -> Any:
         Runs the validations and returns the Param's final value.
         May raise ValueError on failed validations, or TypeError
         if no value is passed and no value already exists.
+        We first check that value is json-serializable; if not, warn.
+        In future release we will require the value to be json-serializable.
 
         :param value: The value to be updated for the Param
         :type value: Any
         :param suppress_exception: To raise an exception or not when the validations fails.
             If true and validations fails, the return value would be None.
         :type suppress_exception: bool
         """
+        try:
+            json.dumps(value)
+        except Exception:
+            warnings.warn(
+                "The use of non-json-serializable params is deprecated and will be removed in "
+                "a future release",
+                DeprecationWarning,
+            )
         final_val = value if value is not NOTSET else self.value
         if isinstance(final_val, ArgNotSet):
             if suppress_exception:
                 return None
-            raise TypeError("No value passed and Param has no default value")
+            raise ParamValidationError("No value passed and Param has no default value")
         try:
             jsonschema.validate(final_val, self.schema, format_checker=FormatChecker())
         except ValidationError as err:
             if suppress_exception:
                 return None
-            raise ValueError(err) from None
+            raise ParamValidationError(err) from None
         self.value = final_val
         return final_val
 
@@ -175,8 +160,8 @@ def __setitem__(self, key: str, value: Any) -> None:
             param = self.__dict[key]
             try:
                 param.resolve(value=value, suppress_exception=self.suppress_exception)
-            except ValueError as ve:
-                raise ValueError(f'Invalid input for param {key}: {ve}') from None
+            except ParamValidationError as ve:
+                raise ParamValidationError(f'Invalid input for param {key}: {ve}') from None
         else:
             # if the key isn't there already and if the value isn't of Param type create a new Param object
             param = Param(value)
@@ -219,8 +204,8 @@ def validate(self) -> Dict[str, Any]:
         try:
             for k, v in self.items():
                 resolved_dict[k] = v.resolve(suppress_exception=self.suppress_exception)
-        except ValueError as ve:
-            raise ValueError(f'Invalid input for param {k}: {ve}') from None
+        except ParamValidationError as ve:
+            raise ParamValidationError(f'Invalid input for param {k}: {ve}') from None
 
         return resolved_dict
 

tests/dags/test_invalid_param.py

@@ -0,0 +1,44 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from datetime import datetime
+
+from airflow import DAG
+from airflow.models.param import Param
+from airflow.operators.python import PythonOperator
+
+with DAG(
+    "test_invalid_param",
+    start_date=datetime(2021, 1, 1),
+    schedule_interval="@once",
+    params={
+        # a mandatory str param
+        "str_param": Param(type="string", minLength=2, maxLength=4),
+    },
+) as the_dag:
+
+    def print_these(*params):
+        for param in params:
+            print(param)
+
+    PythonOperator(
+        task_id="ref_params",
+        python_callable=print_these,
+        op_args=[
+            "{{ params.str_param }}",
+        ],
+    )

tests/jobs/test_scheduler_job.py

@@ -2362,6 +2362,7 @@ def test_list_py_file_paths(self):
             'test_invalid_cron.py',
             'test_zip_invalid_cron.zip',
             'test_ignore_this.py',
+            'test_invalid_param.py',
         }
         for root, _, files in os.walk(TEST_DAG_FOLDER):
             for file_name in files:

tests/models/test_dag.py

@@ -42,7 +42,7 @@
 from airflow import models, settings
 from airflow.configuration import conf
 from airflow.decorators import task as task_decorator
-from airflow.exceptions import AirflowException, DuplicateTaskIdFound
+from airflow.exceptions import AirflowException, DuplicateTaskIdFound, ParamValidationError
 from airflow.models import DAG, DagModel, DagRun, DagTag, TaskFail, TaskInstance as TI
 from airflow.models.baseoperator import BaseOperator
 from airflow.models.dag import dag as dag_decorator
@@ -1861,15 +1861,17 @@ def test_replace_outdated_access_control_actions(self):
 
     def test_validate_params_on_trigger_dag(self):
         dag = models.DAG('dummy-dag', schedule_interval=None, params={'param1': Param(type="string")})
-        with pytest.raises(TypeError, match="No value passed and Param has no default value"):
+        with pytest.raises(ParamValidationError, match="No value passed and Param has no default value"):
             dag.create_dagrun(
                 run_id="test_dagrun_missing_param",
                 state=State.RUNNING,
                 execution_date=TEST_DATE,
             )
 
         dag = models.DAG('dummy-dag', schedule_interval=None, params={'param1': Param(type="string")})
-        with pytest.raises(ValueError, match="Invalid input for param param1: None is not of type 'string'"):
+        with pytest.raises(
+            ParamValidationError, match="Invalid input for param param1: None is not of type 'string'"
+        ):
             dag.create_dagrun(
                 run_id="test_dagrun_missing_param",
                 state=State.RUNNING,

tests/models/test_dagbag.py

@@ -222,6 +222,19 @@ def test_process_file_cron_validity_check(self):
         assert len(dagbag.import_errors) == len(invalid_dag_files)
         assert len(dagbag.dags) == 0
 
+    def test_process_file_invalid_param_check(self):
+        """
+        test if an invalid param in the dag param can be identified
+        """
+        invalid_dag_files = ["test_invalid_param.py"]
+        dagbag = models.DagBag(dag_folder=self.empty_dir, include_examples=False)
+
+        assert len(dagbag.import_errors) == 0
+        for file in invalid_dag_files:
+            dagbag.process_file(os.path.join(TEST_DAGS_FOLDER, file))
+        assert len(dagbag.import_errors) == len(invalid_dag_files)
+        assert len(dagbag.dags) == 0
+
     @patch.object(DagModel, 'get_current')
     def test_get_dag_without_refresh(self, mock_dagmodel):
         """

tests/models/test_param.py

@@ -20,6 +20,7 @@
 import pytest
 
 from airflow.decorators import task
+from airflow.exceptions import ParamValidationError
 from airflow.models.param import Param, ParamsDict
 from airflow.utils import timezone
 from airflow.utils.types import DagRunType
@@ -36,7 +37,7 @@ def test_param_without_schema(self):
 
     def test_null_param(self):
         p = Param()
-        with pytest.raises(TypeError, match='No value passed and Param has no default value'):
+        with pytest.raises(ParamValidationError, match='No value passed and Param has no default value'):
             p.resolve()
         assert p.resolve(None) is None
 
@@ -48,7 +49,7 @@ def test_null_param(self):
         p = Param(None, type='null')
         assert p.resolve() is None
         assert p.resolve(None) is None
-        with pytest.raises(ValueError):
+        with pytest.raises(ParamValidationError):
             p.resolve('test')
 
     def test_string_param(self):
@@ -62,9 +63,9 @@ def test_string_param(self):
         assert p.resolve() == '10.0.0.0'
 
         p = Param(type='string')
-        with pytest.raises(ValueError):
+        with pytest.raises(ParamValidationError):
             p.resolve(None)
-        with pytest.raises(TypeError, match='No value passed and Param has no default value'):
+        with pytest.raises(ParamValidationError, match='No value passed and Param has no default value'):
             p.resolve()
 
     def test_int_param(self):
@@ -74,7 +75,7 @@ def test_int_param(self):
         p = Param(type='integer', minimum=0, maximum=10)
         assert p.resolve(value=5) == 5
 
-        with pytest.raises(ValueError):
+        with pytest.raises(ParamValidationError):
             p.resolve(value=20)
 
     def test_number_param(self):
@@ -84,8 +85,9 @@ def test_number_param(self):
         p = Param(1.0, type='number')
         assert p.resolve() == 1.0
 
-        with pytest.raises(ValueError):
+        with pytest.raises(ParamValidationError):
             p = Param('42', type='number')
+            p.resolve()
 
     def test_list_param(self):
         p = Param([1, 2], type='array')
@@ -124,8 +126,9 @@ def __init__(self, path: str):
         p = S3Param("s3://my_bucket/my_path")
         assert p.resolve() == "s3://my_bucket/my_path"
 
-        with pytest.raises(ValueError):
+        with pytest.raises(ParamValidationError):
             p = S3Param("file://not_valid/s3_path")
+            p.resolve()
 
     def test_value_saved(self):
         p = Param("hello", type="string")
@@ -173,7 +176,7 @@ def test_params_dict(self):
         pd3.validate()
 
         # Update the ParamsDict
-        with pytest.raises(ValueError, match=r'Invalid input for param key: 1 is not'):
+        with pytest.raises(ParamValidationError, match=r'Invalid input for param key: 1 is not'):
             pd3['key'] = 1
 
         # Should not raise an error as suppress_exception is True
@@ -186,7 +189,7 @@ def test_update(self):
         pd.update({'key': 'a'})
         internal_value = pd.get_param('key')
         assert isinstance(internal_value, Param)
-        with pytest.raises(ValueError, match=r'Invalid input for param key: 1 is not'):
+        with pytest.raises(ParamValidationError, match=r'Invalid input for param key: 1 is not'):
             pd.update({'key': 1})
 
 
@@ -270,4 +273,5 @@ def return_num(num):
 
     def test_param_non_json_serializable(self):
         with pytest.warns(DeprecationWarning, match='The use of non-json-serializable params is deprecated'):
-            Param(default={0, 1, 2})
+            p = Param(default={0, 1, 2})
+            p.resolve()