'''
The airflow DAG scaffold for dagster_examples.toys.optional_outputs.optional_outputs

Note that this docstring must contain the strings "airflow" and "DAG" for
Airflow to properly detect it as a DAG
See: http://bit.ly/307VMum
'''
import datetime
import yaml

from dagster_airflow.factory import make_airflow_dag_kubernetized


ENVIRONMENT = '''
loggers:
  console:
    config:
      log_level: DEBUG

solids:
  multiply_the_word:
    inputs:
      word:
        value: bar
    config:
      factor: 2

storage:
  s3:
    config:
      s3_bucket: dagster-airflow-scratch

'''

DEFAULT_ARGS = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime.datetime(2019, 5, 7),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
}

dag, tasks = make_airflow_dag_kubernetized(
    module_name='dagster_airflow_tests.test_project.dagster_airflow_demo',
    pipeline_name='demo_pipeline',
    environment_dict=yaml.load(ENVIRONMENT),
    dag_kwargs={'default_args': DEFAULT_ARGS, 'max_active_runs': 1},
    image='dagster-airflow-demo',
    namespace='default',
)