Basic Airflow DAG Blueprint

icecream_sunday_dag_def.py

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.python_operator import PythonOperator
import random
from pprint import pprint

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2019, 1, 1),
    'email': ['[email protected]'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}
dag = DAG('ice_cream_sundae', default_args=default_args, schedule_interval=None)

sundae_choices = {
    'cones': ['cup', 'waffle'],
    'toppings': ['rainbow sprinkles', 'chocolate sprinkles', 'm&ms', 'gummy bears', 'chocolate sauce',
                 'strawberry sauce', 'caramel sauce', 'marshmallow'],
    'ice_cream_flavors': ['vanilla', 'chocolate', 'strawberry', 'salted caramel', 'cotton candy'],
}


def choose_cone(ds, **kwargs):
    pass


choose_cone_op = PythonOperator(
    task_id='choose_cone_task',
    python_callable=choose_cone,
    provide_context=True,
    dag=dag
)


def choose_icecream_flavor(ds, **kwargs):
    pass


choose_icecream_flavor_op = PythonOperator(
    task_id='choose_icecream_flavor_task',
    provide_context=True,
    dag=dag,
    python_callable=choose_icecream_flavor,
)


def choose_toppings(ds, **kwargs):
    pass


choose_toppings_op = PythonOperator(
    task_id='choose_toppings_task',
    provide_context=True,
    dag=dag,
    python_callable=choose_toppings,
)


def make_icecream_sundae(ds, **kwargs):
    pass


make_icecream_sundae_op = PythonOperator(
    task_id='make_icecream_sundae_task',
    provide_context=True,
    dag=dag,
    python_callable=make_icecream_sundae,
)

choose_icecream_flavor_op.set_upstream(choose_cone_op)
choose_toppings_op.set_upstream(choose_icecream_flavor_op)
make_icecream_sundae_op.set_upstream(choose_toppings_op)