Rendering options

[rory-data]

Hi team.

This is more of a question for my own understanding, but why is Blueprint trying to support Python 3.8 and 3.9? With them being end of life (or about to be) and Airflow having dropped support for them, I'm curious.

For context, I've been working through options for how to extend this to have the option for dynamically generating DAG files on build instead of at runtime. Unsure if this is a desired feature for this project, but the factory pattern can make support/debugging more complex. Therefore, my shop has a requirement to be able to see the fully rendered DAG code in the Airflow UI code view, and finding this project seemed fortuitous timing.

There appear to be two options for this:

1. Store the DAG blueprint as a *.py.j2 file and have that accessible by both render() and a new render_to_file() method, or have the DAG definition that's currently in render() be a Jinja2-compatible text block. Simplest approach but may require more care when developing new blueprints.
2. Have render() return the DAG object as it does now, but add functionality in core.py to write that out to file. Cleaner on the blueprint side, but much more complex to implement and maintain.

They're both doable, but the lower Python compatibility requirements add more complexity for what may be little gain, thus my original question. Of course, if the rendering to file is not an intended feature, then it's likely a moot point 😄

Cheers
Rory

[jlaneve]

hey @rory-data - thanks for the question! I'm good to drop the support for 3.8 and 3.9. I figured, starting out, might as well support everything we can and then drop support for what we don't need 🙂

fwiw I do like the idea of a render_to_file. We've tried a couple iterations of option 2 with varying success, so I'd rather not go down that road until it becomes obvious we need to. I wonder if we could require either render or render_to_file - in your case, it sounds like you'd want to define blueprints that just write files, correct?

I'll go ahead and drop support for 3.8/3.9. Let me know if you want to take a first pass at generating these on build time

[rory-data]

Hey @jlaneve. I've finally had a chance to start having a look at this... I suspect the easiest approach would be to fold in some features of Orbiter to save reinventing the wheel as that already tackles some of the parsing complexity for us.

I was thinking back to my team's original problem statement of wanting to be able to see the DAG code in the UI and realised a relatively simple solution. If the blueprint is moved to the dags directory and some other components are moved out, we get as close as we're going to for a UI-readable solution (as per the below). However, while it does work with the Blueprint code as it stands now, it's a bit of a design shift of where the template and supporting code sits.

"""
Different approach for the example of using Blueprint's Python API to create DAGs dynamically.

This blueprint template is in the DAGs directory so is the object parsed by Airflow and represented
as-is in the UI Code view. The blueprint config is moved out to make this cleaner to read in the UI
as are the build and globals registration methods.
"""

from datetime import datetime, timedelta, timezone

from blueprint import Blueprint

from include.daily_etl_config import DailyETLConfig
from include.proto_utils import build_dags


class DailyETL(Blueprint[DailyETLConfig]):
    """Daily ETL job that moves data between tables with configurable scheduling."""

    def render(self, config: DailyETLConfig):
        from airflow import DAG
        from airflow.operators.bash import BashOperator
        from airflow.operators.python import PythonOperator

        default_args = {
            "owner": "data-team",
            "retries": config.retries,
            "retry_delay": timedelta(minutes=5),
            "email_on_failure": False,
        }
        dag = DAG(
            dag_id=config.job_id,
            default_args=default_args,
            description=f"ETL from {config.source_table} to {config.target_table}",
            schedule=config.schedule,
            start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
            catchup=False,
            tags=["etl", "blueprint"],
        )
        with dag:
            check_source = BashOperator(
                task_id="check_source_data",
                bash_command=f'echo "Checking if {config.source_table} has data..."',
            )

            def extract_transform(**_):
                print(f"Extracting data from {config.source_table}")
                print("Applying transformations...")
                print(f"Preparing to load into {config.target_table}")
                return {"records_processed": 1000}

            etl_task = PythonOperator(
                task_id="extract_transform",
                python_callable=extract_transform,
            )
            load_data = BashOperator(
                task_id="load_data",
                bash_command=f'echo "Loading data into {config.target_table}..."',
            )
            quality_check = BashOperator(
                task_id="data_quality_check",
                bash_command=f'echo "Running quality checks on {config.target_table}..."',
            )
            check_source >> etl_task >> load_data >> quality_check
        return dag

param = {
    "job_id": "all_customer-etl",
    "source_table": "raw.all_customers",
    "target_table": "staging.all_customers_clean",
    "schedule": "@daily",
    "retries": 3,}

build_dags(template=DailyETL, param=param)

[jlaneve]

hey @rory-data apologies for the delay here, got caught up at Airflow Summit last week - I do think this approach is pretty interesting! This way, to your point, the code gets shown in the Airflow UI. We can also turn the YAML into a rendered template (we do this in Cosmos) so that DAG authors can see both a) the template generation code above, and b) the actual config specified for their current DAG. Thoughts?