feat: kedro-airflow DAG kwarg configuration

[sbrugman]

Description

Implementation for #229

Development notes

Feature implementation for configuring kedro Airflow DAGs via airflow.yml.
The arguments can be configured per pipeline.

The arguments are rendered in the template. Initially I attempted specifying variables to change directly in the yaml file. However, rendering the variables in Python requires dealing with the data types of each argument (e.g. datetime, dict, string, bool ...). This added a lot of complexity to the plugin/template. In practice, the user can take care of this simply by providing a (small) template.

In the setup, the user can update the jinja template for new arguments, then can use airflow.yml:

airflow.yml:

# Global
default:
   max_active_runs: 3
   schedule_interval: "@once"
   start_month: 1
   owner: "airflow"

# Per Pipeline
data_science:
   start_month: 2
   owner: "airflow-ds"   

Additionally, parameters can be passed with the --params CLI option.

The configuration pattern is also customizable.

Tests are present.

Checklist

• Opened this PR as a 'Draft Pull Request' if it is work-in-progress
• Updated the documentation to reflect the code changes
• Added a description of this change in the relevant RELEASE.md file
• Added tests to cover my changes

[sbrugman]

@noklam @astrojuanlu @deepyaman Could anyone please have a brief review of this PR and #241? Conceptually they are relatively small changes, so they should not take much time to review. They do enable us and other users to switch from kedro-airlfow-k8s to this official plugin!

Feel free to be critical and brief. Happy to put in the refactor work to get these features supported!

[noklam]

I've left some high-level comments, I didn't review the details of the implementation yet.

Can you provide some steps how can I test with this new feature? I also expect there will be some unit tests or e2e test to make sure it works.

[sbrugman]

@noklam Could you please have another look? The implementation is now simpler and has proper tests.

In addition to the configuration this PR also:

• Adds help messages to the CLI arguments (contributing to Provide more helpful message for kedro airflow create -h #244)
• Handles the case when a pipeline is specified but does not exists more gracefully
• Adds documentation on how a user would use (native) Airflow parameters
• Handles whitespaces in the resulting DAG better

[luizvbo]

I left some minor comments.

[luizvbo]

I ran the plugin on a kedro starter project and the result seems correct.

[sbrugman]

Thanks @luizvbo! Processed your comments.

[astrojuanlu]

Happy to give this a deeper look when the linting and test failures are addressed

[sbrugman]

A from __future__ import annotations import was missing, causing the lints/tests to fail on older Python versions. Fixed that.

[astrojuanlu]

Nice, thanks - the last thing is some failures on Windows:

    def test_create_airflow_dag_nonexistent_pipeline(cli_runner, metadata):
        """Test executing with a non-existing pipeline"""
        command = ["airflow", "create", "--pipeline", "de"]
        result = cli_runner.invoke(commands, command, obj=metadata)
        assert result.exit_code == 1
>       assert (
            "kedro.framework.cli.utils.KedroCliError: Pipeline de not found."
            in result.stdout
        )
E       assert 'kedro.framework.cli.utils.KedroCliError: Pipeline de not found.' in ''

[ankatiyar]

Thanks for this @sbrugman! I've left some comments about it's compatibility with OmegaConfigLoader which needs to be addressed! :)

[ankatiyar]

Did another round of review and I tested it with a starter project. Some minor comments, mostly the README/Release notes. I think this is very close to the finish line, I'll add more reviewers for approval. 💯

[ankatiyar]

@sbrugman I'm sorry, I believe committing suggestions directly from here don't get signed off making the DCO bot fail. 😅

[sbrugman]

Let me know if there is anything else there is to be done on our side to help!
Quite eager to start using this feature in production.

[noklam]

@sbrugman Sorry for blocking this PR for so long, I have missed this notification. If you're on Slack feel free to ping me directly.

I made a commit to fix some minor thing about code style. I have tested both --params and airflow.yml, works pretty well for me.

One thing that is less intuitive to me is the airflow.yml parameters doesn't work the same as Kedro's params, as the top level key is pipeline. i.e. if you want to overide a subkey pipeline.key=3, you will do kedro airflow create -p pipeline --params key=3 instead of kedro airflow create --params pipeline.key=3. I think this is fine but just want to make a note about this.

In addition, it may be helpful to create a kedro airflow init to generate the airflow.yml with default argument at some point, I've manually test by creating this file for the sake of review.

Excellent work⭐️ We should follow up with a release after this PR.

[noklam]

👍🏼 I like this.

[merelcht]

I left a super minor comment, but mainly came here to say thank you for this great contribution @sbrugman ! 🤩 I haven't been able to try this yet, but the code looks good and thanks for adding docs as well 👍

[ankatiyar]

Thank you @sbrugman for this! I hope you don't mind that I made some commits to resolve conflicts and incorporate the suggestions to the docs from the reviews. Very happy about getting this merged in! 🥳 💯

[sbrugman]

Thank you everyone for the collaboration! This feature allows us to make an important step in our usage of scheduled Kedro nodes in many real-world pipelines 👌

Great to see that this was merged during my holidays 🌴

[noklam]

@sbrugman Thank you for your contribution and hope you have enjoyed your holiday! We have already done a release so please go ahead and use it!