@provide_session
    def set_dag_runs_state(
            self, state=State.RUNNING, session=None):
        drs = session.query(DagModel).filter_by(dag_id=self.dag_id).all()
        dirty_ids = []