"""
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta


default_args = {
    'owner': 'lyin',
    'depends_on_past': False,
    'start_date': datetime(2016, 10, 15),
    'email': ['leon@theorchard.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}

dag = DAG(
    'tutorial_v1', default_args=default_args, schedule_interval='@daily')

# # t1, t2 and t3 are examples of tasks created by instantiating operators
# t1 = BashOperator(
#     task_id='print_date',
#     bash_command='date',
#     dag=dag)

# t2 = BashOperator(
#     task_id='sleep',
#     bash_command='sleep 5',
#     retries=3,
#     dag=dag)

# templated_command = """
#     {% for i in range(5) %}
#         echo "{{ ds }}"
#         echo "{{ macros.ds_add(ds, 7)}}"
#         echo "{{ params.my_param }}"
#     {% endfor %}
# """

# t3 = BashOperator(
#     task_id='templated',
#     bash_command=templated_command,
#     params={'my_param': dag.default_args['start_date']},
#     dag=dag)

def print_num(**kwargs):
    print(vars(kwargs.dag))
    print(kwargs.dag.default_args.keys())
    ti = kwargs['ti']
    print( kwargs['ti'].xcom_pull(key='blah', task_ids='bootstrap') )

def bootstrap(**kwargs):
    kwargs['ti'].xcom_push(key='blah', value=kwargs['ds'])
    kwargs['ti'].xcom_push(key='blah2', value=kwargs['execution_date'])

t1 = PythonOperator(
    task_id = 'bootstrap',
    python_callable = bootstrap,
    provide_context= True,
    dag = dag
)

t2 = PythonOperator(
    task_id = 'print_context',
    python_callable = print_num,
    provide_context= True,
    dag = dag
)

t2.set_upstream(t1)