Nested parameters and JSON Schema with arrays for DAG Run configurations

[varunkvv]

I'm unsure if this is a bug or I'm doing something wrong. This is on Airflow v2.3.0

To set the context, I'm trying to create a DAG using JSON Schema / Param:

dag = DAG(
    dag_id="example_dag",
    start_date=datetime(2021, 1, 1),
    catchup=False,
    doc_md=__doc__,
    schedule_interval=None,
    params={
        "accounts": Param(
            [{"name": "account_name_1", "country": "usa"}],
            type="array",
            items={
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "country": {"type": "string"},
                },
            },
        ),
    }
)

In particular, the 'accounts' field takes an array that can take an arbitrary number of objects, but with a fixed schema.

When I go to the Airflow's "Trigger DAG w/ config" page the default shows up like this:

{
    "accounts": [
        {
            "__var": {
                "name": "account_name_1",
                "country": "usa"
            },
            "__type": "dict"
        }
    ]
}

This seems to get passed verbatim to the DAG itself, which is not correct.

What does seem to mostly work is the JSON Schema itself. For example, if I put a wrong type for accounts.country it gives the correct error message. So the issue mostly seems to be about rendering the default parameters in the UI (and possibly the dependent backends).

---

EDIT: the issue with required below was on my end, can be ignored.
On a secondary note, it looks like adding required modifiers to the JSON Schema completely breaks the DAG parsing. The DAG does not show up, and it does not provide an error message on the top. I can file an issue for this if deemed appropriate.

[varunkvv]

An update: I changed the params to include schema field directly like this, and the JSON Schema bit has begun working consistently.

DAG(
    dag_id="some_id",
    start_date=datetime(2021, 1, 1),
    catchup=False,
    doc_md=__doc__,
    schedule_interval=None,
    params={
        "accounts": Param(
            [{'name': 'account_name_1', 'country': 'usa'}],
            schema =  {
                "type": "array",
                "minItems": 1,
                "items": {
                    "type": "object",
                    "default": {"name": "account_name_1", "country": "usa"},
                    "properties": {
                        "name": {"type": "string"},
                        "region": {"type": "string"},
                    },
                    "required": ["name", "region"]
                },
            }
        ),
    }
)

However, now the default value is not used at all. In the DAG Run trigger page, I now see this:

{
    "accounts": null
}

This is overall better, but I'm wondering why I can't get both the default value and consistent JSON Schema validation for this case.

[potiuk]

Maybe @msumit can help ?

[technopahadi]

It might be an issue with the pickled/serialized DAG. I have a setup with Scheduler and web server running in different pods. From the scheduler if I run the following code (with the example given above), things seem fine (because the dag bag is constructed from the files I think):

>>> from airflow.models import DagBag
>>> bag = DagBag()
>>> dag = bag.get_dag('dummy_dag')
>>> for k, v in dag.params.items():
...     print(k, v.resolve())
...
accounts [{'name': 'account_name_1', 'country': 'usa'}]

But in the web server process (in Flask app context), this is what I get:

[{'__var': {'name': 'account_name_1', 'country': 'usa'}, '__type': 'dict'}]

Just for ref, the code I used in Flask's context is this:

from airflow.www.app import create_app

app = create_app()

with app.app_context():
    from airflow.utils.airflow_flask_app import get_airflow_app

    dag = get_airflow_app().dag_bag.get_dag('dummy_dag')
    for k, v in dag.params.items():
        print(k, v.resolve())

Because the web server's code actually does v.resolve(suppress_exception=True) in views.py the returned value is None or null in the UI. Since the response that is gotten above doesn't validate with the schema, exception is thrown in Params class.

I think this a bug. @varunkvv @potiuk Would you want to file a bug or should I go ahead?

[potiuk]

go ahead and file it.

[technopahadi]

#27228

[ppseverin]

I don't know if anyone managed to solve this, but I tried with the kwarg "schema" and it kinda worked. Now the default value always includes "__type": "dict" and "__var":{.... If I remove all those it runs just fine.

The other thing I noticed is that the title, type and section (and any other keyword except for description and default) are ignored.

I went to the param.py code and I found that when schema is provided, every other keyword is ignored, except for description and default:

class Param:
    """
    Class to hold the default value of a Param and rule set to do the validations.

    Without the rule set it always validates and returns the default value.

    :param default: The value this Param object holds
    :param description: Optional help text for the Param
    :param schema: The validation schema of the Param, if not given then all kwargs except
        default & description will form the schema
    """

Is there any way to fix this without using the schema keyword?