text: from datetime import datetime, timedelta
  import os
  from airflow import DAG
  from airflow.operators.dummy_operator import DummyOperator
  from airflow.operators import (StageToRedshiftOperator, LoadFactOperator,
                                  LoadDimensionOperator, DataQualityOperator)
  from helpers import SqlQueries
  
  default_args = {
      'owner': 'udacity',
      'start_date': datetime(2019, 1, 12),
      'retries': 3,
      'retry_delay': timedelta(minutes=5),
      'email_on_retry': False
  }
  
  dag = DAG('udac_example_dag',
            default_args=default_args,
            description='Load and transform data in Redshift with Airflow',
            schedule_interval='0 * * * *',
            catchup=False
          )
  
  start_operator = DummyOperator(task_id='Begin_execution',  dag=dag)
  
  stage_events_to_redshift = StageToRedshiftOperator(
      task_id='Stage_events',
      dag=dag,
      table="staging_events",
      redshift_conn_id="redshift",
      aws_credentials_id="aws_credentials",
      s3_bucket="udacity-dend",
      s3_key="log_data",
      json_path="s3://udacity-dend/log_json_path.json",
      extra_params="format as json 's3://udacity-dend/log_json_path.json'"
  )
  
  stage_songs_to_redshift = StageToRedshiftOperator(
      task_id='Stage_songs',
      dag=dag,
      redshift_conn_id="redshift",
      aws_credentials_id="aws_credentials",
      table="staging_songs",
      s3_bucket='udacity-dend',
      s3_key="song_data",
      extra_params="json 'auto' compupdate off region 'us-west-2'",
  )
  
  load_songplays_table = LoadFactOperator(
      task_id='Load_songplays_fact_table',
      dag=dag,
      redshift_conn_id="redshift",
      table="songplays",
      sql_source=SqlQueries.songplay_table_insert,
  )
  
  load_user_dimension_table = LoadDimensionOperator(
      task_id='Load_user_dim_table',
      dag=dag,
      redshift_conn_id="redshift",
      table="users",
      sql_source=SqlQueries.user_table_insert,
  )
  
  load_song_dimension_table = LoadDimensionOperator(
      task_id='Load_song_dim_table',
      dag=dag,
      redshift_conn_id="redshift",
      table="songs",
      sql_source=SqlQueries.song_table_insert,
  )
  
  load_artist_dimension_table = LoadDimensionOperator(
      task_id='Load_artist_dim_table',
      dag=dag,
      redshift_conn_id="redshift",
      table="artists",
      sql_source=SqlQueries.artist_table_insert,
  )
  
  load_time_dimension_table = LoadDimensionOperator(
      task_id='Load_time_dim_table',
      dag=dag,
      redshift_conn_id="redshift",
      table="time",
      sql_source=SqlQueries.time_table_insert,
  )
  
  run_quality_checks = DataQualityOperator(
      task_id='Run_data_quality_checks',
      dag=dag,
      redshift_conn_id="redshift",
      tables=['songplays', 'users', 'songs', 'artists', 'time'],
  )
  
  end_operator = DummyOperator(task_id='Stop_execution',  dag=dag)
  
  
  # Configuring DAG flow
  start_operator >> stage_events_to_redshift
  start_operator >> stage_songs_to_redshift
  
  stage_events_to_redshift >> load_songplays_table
  stage_songs_to_redshift >> load_songplays_table
  
  load_songplays_table >> load_song_dimension_table
  load_songplays_table >> load_user_dimension_table
  load_songplays_table >> load_artist_dimension_table
  load_songplays_table >> load_time_dimension_table
  
  load_song_dimension_table >> run_quality_checks
  load_user_dimension_table >> run_quality_checks
  load_artist_dimension_table >> run_quality_checks
  load_time_dimension_table >> run_quality_checks
  
  run_quality_checks >> end_operator