Missing required parameter shows a stacktrace instead of a validation error

[MatrixManAtYrService]

Apache Airflow version

2.2.3 (latest released)

What happened

I wanted to improve the Params concepts doc to show how one can reference the params from a task. While doing so, I tried to run that DAG by clicking "Trigger" (i.e. I didn't opt to modify the params first). Then I saw this:

Oops.
Something bad has happened.
...
Python version: 3.9.9
Airflow version: 2.2.3+astro.1
Node: df9204fff6e6
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/airflow/models/param.py", line 62, in __init__
    jsonschema.validate(self.value, self.schema, format_checker=FormatChecker())
  File "/usr/local/lib/python3.9/site-packages/jsonschema/validators.py", line 934, in validate
    raise error
jsonschema.exceptions.ValidationError: 'NoValueSentinel' is too long

Failed validating 'maxLength' in schema:
    {'maxLength': 4, 'minLength': 2, 'type': 'string'}

On instance:
    'NoValueSentinel'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python3.9/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/auth.py", line 51, in decorated
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/decorators.py", line 72, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/views.py", line 1650, in trigger
    dag = current_app.dag_bag.get_dag(dag_id)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py", line 186, in get_dag
    self._add_dag_from_db(dag_id=dag_id, session=session)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py", line 261, in _add_dag_from_db
    dag = row.dag
  File "/usr/local/lib/python3.9/site-packages/airflow/models/serialized_dag.py", line 180, in dag
    dag = SerializedDAG.from_dict(self.data)  # type: Any
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 947, in from_dict
    return cls.deserialize_dag(serialized_obj['dag'])
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 861, in deserialize_dag
    v = {task["task_id"]: SerializedBaseOperator.deserialize_operator(task) for task in v}
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 861, in <dictcomp>
    v = {task["task_id"]: SerializedBaseOperator.deserialize_operator(task) for task in v}
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 641, in deserialize_operator
    v = cls._deserialize_params_dict(v)
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 459, in _deserialize_params_dict
    op_params[k] = cls._deserialize_param(v)
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 439, in _deserialize_param
    return class_(**kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/param.py", line 64, in __init__
    raise ValueError(err)
ValueError: 'NoValueSentinel' is too long

Failed validating 'maxLength' in schema:
    {'maxLength': 4, 'minLength': 2, 'type': 'string'}

On instance:
    'NoValueSentinel'

I started with the dag in the docs, but ended up making some tweaks. Here's how it was when I saw the error.

from airflow import DAG
from airflow.models.param import Param
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    "my_dag",
    start_date=datetime(1970, 1, 1),
    schedule_interval=None,
    params={
        # a int param with default value
        "int_param": Param(10, type="integer", minimum=0, maximum=20),

        # a mandatory str param
        "str_param": Param(type="string", minLength=2, maxLength=4),

        # a param which can be None as well
        "dummy": Param(type=["null", "number", "string"]),

        # no data or type validations
        "old": "old_way_of_passing",

        # no data or type validations
        "simple": Param("im_just_like_old_param"),
        "email": Param(
            default="example@example.com",
            type="string",
            format="idn-email",
            minLength=5,
            maxLength=255,
        ),
    },
) as the_dag:

    def print_these(*params):
        for param in params:
            print(param)

    PythonOperator(
        task_id="ref_params",
        python_callable=print_these,
        op_args=[
            # you can modify them in jinja templates
            "{{ params.int_param + 10 }}",
            # or just leave them as-is
            "{{ params.str_param }}",
            "{{ params.dummy }}",
            "{{ params.old }}",
            "{{ params.simple }}",
            "{{ params.email }}",
        ],
    )

What you expected to happen

If there's a required parameter, and I try to trigger a dagrun, I should get a friendly warning explaining what I've done wrong.

How to reproduce

Run the dag above

Operating System

docker / debian

Versions of Apache Airflow Providers

n/a

Deployment

Astronomer

Deployment details

astro dev start

Dockerfile:

FROM quay.io/astronomer/ap-airflow:2.2.3-onbuild

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[kaxil]

cc @ephraimbuddy Something for you to look at when you are back

[uranusjr]

Looks like the same issue as #20442.