from airflow import DAG
from airflow.utils.dates import days_ago

from anyway_etl_airflow.operators.cli_bash_operator import CliBashOperator


dag_kwargs = dict(
    default_args={
        'owner': 'airflow',
    },
    schedule_interval='@weekly',
    catchup=False,
    start_date=days_ago(2),
)


with DAG('cbs', **dag_kwargs,
         description='by default imports emails and processes data for (current year - 1). '
                     'For back-fill do a manual run with following example json: '
                     '{"load_start_year": 2019}') as cbs_dag:
    CliBashOperator(
        'anyway-etl cbs import-emails',
        skip_if=lambda context: context['dag_run'].conf.get('load_start_year'),
        task_id='import-emails'
    ) >> CliBashOperator(
        'anyway-etl cbs process-files',
        skip_if=lambda context: context['dag_run'].conf.get('load_start_year'),
        task_id='process-files'
    ) >> [
        # for local development you can use the following command to parse all types sequentially:
        #   anyway-etl cbs parse-all
        CliBashOperator(
            'anyway-etl cbs parse-accidents'
            '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
            task_id='parse-accidents'
        ),
        CliBashOperator(
            'anyway-etl cbs parse-involved'
            '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
            task_id='parse-involved'
        ),
        CliBashOperator(
            'anyway-etl cbs parse-vehicles'
            '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
            task_id='parse-vehicles'
        ),
    ] >> CliBashOperator(
        'anyway-etl cbs import-to-datastore'
        '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
        task_id='import-to-datastore'
    ) >> CliBashOperator(
        'anyway-etl cbs check-data-in-datastore',
        task_id='check-data-in-datastore'
    )

from airflow import DAG
from airflow.utils.dates import days_ago

from anyway_etl_airflow.operators.cli_bash_operator import CliBashOperator


dag_kwargs = dict(
    default_args={
        'owner': 'airflow',
    },
    schedule_interval='@weekly',
    catchup=False,
    start_date=days_ago(2),
)


with DAG('cbs', **dag_kwargs,
         description='by default imports emails and processes data for (current year - 1). '
                     'For back-fill do a manual run with following example json: '
                     '{"load_start_year": 2019}') as cbs_dag:
    CliBashOperator(
        'anyway-etl cbs import-emails',
        skip_if=lambda context: context['dag_run'].conf.get('load_start_year'),
        task_id='import-emails'
    ) >> CliBashOperator(
        'anyway-etl cbs process-files',
        skip_if=lambda context: context['dag_run'].conf.get('load_start_year'),
        task_id='process-files'
    ) >> [
        # for local development you can use the following command to parse all types sequentially:
        #   anyway-etl cbs parse-all
        CliBashOperator(
            'anyway-etl cbs parse-accidents'
            '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
            task_id='parse-accidents'
        ),
        CliBashOperator(
            'anyway-etl cbs parse-involved'
            '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
            task_id='parse-involved'
        ),
        CliBashOperator(
            'anyway-etl cbs parse-vehicles'
            '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
            task_id='parse-vehicles'
        ),
    ] >> CliBashOperator(
        'anyway-etl cbs import-to-datastore'
        '{% if dag_run.conf.get("load_start_year") %} --load-start-year {{ dag_run.conf["load_start_year"] }}{% endif %}',
        task_id='import-to-datastore'
    ) >> CliBashOperator(
        'anyway-etl cbs check-data-in-datastore',
        task_id='check-data-in-datastore'
    )

# for local development you can use the following command to parse all types sequentially: #   anyway-etl cbs parse-all