__init__.py

__init__.py

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

from .tasks import run_package, send_slack_alert


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2019, 7, 19),
    'email': ['[email protected]'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0
}

dag = DAG('xplenty_mongo', default_args=default_args,)

package_run_config = {
    'cluster': {
        'cluster_type': 'production',
        'nodes': 1,
        'name': 'XPlenty-Mongo',
        'description': 'Cluster for Mongo to Redshift pipeline',
        'terminate_on_idle': True,
        'time_to_idle': 3600
    },
    'package': {
        'variables': {},
        'package_id': 107387
    }
}

run_package_task = PythonOperator(task_id='run_package', python_callable=run_package, dag=dag,
                                  op_kwargs=package_run_config)

send_slack_alert_task = PythonOperator(task_id='send_slack-alert',
                                       python_callable=send_slack_alert,
                                       dag=dag,
                                       trigger_rule="all_done")

run_package_task.set_downstream(send_slack_alert_task)