'Updates the stats for dirty/out-of-sync dags
:param dag_ids: dag_ids to be updated
:type dag_ids: list
:param dirty_only: only updated for marked dirty, defaults to True
:type dirty_only: bool
:param session: db session to use
:type session: Session'

@staticmethod
@provide_session
def update(dag_ids=None, dirty_only=True, session=None):

    try:
        qry = session.query(DagStat)
        if dag_ids:
            qry = qry.filter(DagStat.dag_id.in_(set(dag_ids)))
        if dirty_only:
            qry = qry.filter((DagStat.dirty == True))
        qry = qry.with_for_update().all()
        ids = set([dag_stat.dag_id for dag_stat in qry])
        if (len(ids) == 0):
            session.commit()
            return
        dagstat_states = set(itertools.product(ids, State.dag_states))
        qry = session.query(DagRun.dag_id, DagRun.state, func.count(u'*')).filter(DagRun.dag_id.in_(ids)).group_by(DagRun.dag_id, DagRun.state)
        counts = {(dag_id, state): count for (dag_id, state, count) in qry}
        for (dag_id, state) in dagstat_states:
            count = 0
            if ((dag_id, state) in counts):
                count = counts[(dag_id, state)]
            session.merge(DagStat(dag_id=dag_id, state=state, count=count, dirty=False))
        session.commit()
    except Exception as e:
        session.rollback()
        logging.warning(u'Could    not    update    dag    stat    table')
        logging.exception(e)
