':param dag_id: the dag_id to mark dirty
:param session: database session
:return:'

@staticmethod
@provide_session
def set_dirty(dag_id, session=None):

    DagStat.create(dag_id=dag_id, session=session)
    try:
        stats = session.query(DagStat).filter((DagStat.dag_id == dag_id)).with_for_update().all()
        for stat in stats:
            stat.dirty = True
        session.commit()
    except Exception as e:
        session.rollback()
        logging.warning(u'Could    not    update    dag    stats    for    {}'.format(dag_id))
        logging.exception(e)
