Full Dynamic CAN Airflow

dynamic_can_airflow.py

from collections import defaultdict
from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

from pyseir.run import OneRegionPipeline

from libs.datasets import combined_datasets

default_args = {
    'owner': 'airflow',
}


@dag(default_args=default_args, schedule_interval=None, start_date=days_ago(2))
def produce_dynamic_dag():
    @task()
    def transform(region: str, one_region):
        """
        #### Counts the sub-regions
        """
        OneRegionPipeline.run(one_region)

    # Produce the DAG structure dynamically by reading all the data.
    regions_dataset = combined_datasets.load_us_timeseries_dataset(
    ).get_subset(states=['NY', 'CA'])
    for region, one_region in regions_dataset.iter_one_regions():
        transform(region, one_region)


dynamic_dag = produce_dynamic_dag()