'Finding all tasks that have SLAs defined, and sending alert emails
where needed. New SLA misses are also recorded in the database.
Where assuming that the scheduler runs often, so we only check for
tasks that should have succeeded in the past hour.'

@provide_session
def manage_slas(self, dag, session=None):

    if (not any([ti.sla for ti in dag.tasks])):
        self.logger.info(u'Skipping    SLA    check    for    {}    because    no    tasks    in    DAG    have    SLAs'.format(dag))
        return
    TI = models.TaskInstance
    sq = session.query(TI.task_id, func.max(TI.execution_date).label(u'max_ti')).with_hint(TI, u'USE    INDEX    (PRIMARY)', dialect_name=u'mysql').filter((TI.dag_id == dag.dag_id)).filter((TI.state == State.SUCCESS)).filter(TI.task_id.in_(dag.task_ids)).group_by(TI.task_id).subquery(u'sq')
    max_tis = session.query(TI).filter((TI.dag_id == dag.dag_id), (TI.task_id == sq.c.task_id), (TI.execution_date == sq.c.max_ti)).all()
    ts = datetime.now()
    SlaMiss = models.SlaMiss
    for ti in max_tis:
        task = dag.get_task(ti.task_id)
        dttm = ti.execution_date
        if task.sla:
            dttm = dag.following_schedule(dttm)
            while (dttm < datetime.now()):
                following_schedule = dag.following_schedule(dttm)
                if ((following_schedule + task.sla) < datetime.now()):
                    session.merge(models.SlaMiss(task_id=ti.task_id, dag_id=ti.dag_id, execution_date=dttm, timestamp=ts))
                dttm = dag.following_schedule(dttm)
    session.commit()
    slas = session.query(SlaMiss).filter(or_((SlaMiss.email_sent == False), (SlaMiss.notification_sent == False))).filter((SlaMiss.dag_id == dag.dag_id)).all()
    if slas:
        sla_dates = [sla.execution_date for sla in slas]
        qry = session.query(TI).filter((TI.state != State.SUCCESS)).filter(TI.execution_date.in_(sla_dates)).filter((TI.dag_id == dag.dag_id)).all()
        blocking_tis = []
        for ti in qry:
            if (ti.task_id in dag.task_ids):
                ti.task = dag.get_task(ti.task_id)
                blocking_tis.append(ti)
            else:
                session.delete(ti)
                session.commit()
        task_list = u'\n'.join([((sla.task_id + u'    on    ') + sla.execution_date.isoformat()) for sla in slas])
        blocking_task_list = u'\n'.join([((ti.task_id + u'    on    ') + ti.execution_date.isoformat()) for ti in blocking_tis])
        email_sent = False
        notification_sent = False
        if dag.sla_miss_callback:
            self.logger.info(u'    -------------->    ABOUT    TO    CALL    SLA    MISS    CALL    BACK    ')
            dag.sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis)
            notification_sent = True
        email_content = u"                                                Here's    a    list    of    tasks    thas    missed    their    SLAs:\n                                                <pre><code>{task_list}\n<code></pre>\n                                                Blocking    tasks:\n                                                <pre><code>{blocking_task_list}\n{bug}<code></pre>\n                                                ".format(bug=asciiart.bug, **locals())
        emails = []
        for t in dag.tasks:
            if t.email:
                if isinstance(t.email, basestring):
                    l = [t.email]
                elif isinstance(t.email, (list, tuple)):
                    l = t.email
                for email in l:
                    if (email not in emails):
                        emails.append(email)
        if (emails and len(slas)):
            send_email(emails, (u'[airflow]    SLA    miss    on    DAG=' + dag.dag_id), email_content)
            email_sent = True
            notification_sent = True
        if notification_sent:
            for sla in slas:
                if email_sent:
                    sla.email_sent = True
                sla.notification_sent = True
                session.merge(sla)
        session.commit()
        session.close()
