import sys
import os

from zenyatta.pipelines import initialize_etl_dag
from zenyatta.common import get_dag_defaults, get_zenyatta_connections
from zenyatta.aws import get_aws_metadata
from sqlalchemy import not_, and_
from sqlalchemy.orm.exc import NoResultFound

from airflow import DAG
from airflow.models import Connection

from airflow import settings

PARENT = '../..'
SCRIPT_DIR = os.path.dirname(os.path.realpath(os.path.join(os.getcwd(), os.path.expanduser(__file__))))
sys.path.append(os.path.normpath(os.path.join(SCRIPT_DIR, PARENT)))

connections = get_zenyatta_connections()
conns = []
for conn in connections:
    if conn.host.endswith('rds.amazonaws.com'):
        meta = {'rds': True}
    else:
        meta = {'rds': False, 'wal_aws_creds': get_aws_metadata(conn.conn_id)}
    conns.append((conn.conn_id, conn.conn_id + '-etl', initialize_etl_dag, meta)),

sesh = settings.Session()
for conn_id, dag_id, dag_init_func, params in conns:
    # merge all args
    initial_args = {
        's3_output': get_aws_metadata('s3-output'),
        'sns_output': get_aws_metadata('sns-output'),
    }
    try:
        new_args, dag_args = get_dag_defaults(conn_id)
        initial_args.update(new_args)
        initial_args.update(params)
    except:
        print("no defaults found for: {}".format(conn_id))
        continue
    try:
        connection = sesh.query(Connection).filter(Connection.conn_id == conn_id).one()
        dag = DAG(dag_id, default_args=initial_args, **dag_args)
        # a role being passed will add additional task(s) to the DAG to open up permissions on that role
        if 'role' in initial_args:
            dag_init_func(connection, dag, role=initial_args.get('role'))
        else:
            dag_init_func(connection, dag)
        globals()[dag_id] = dag
    except NoResultFound:
        print("no connection found for " + conn_id)
