start_date=execution_date,
            end_date=execution_date,
            dag_run_state=False,  
            session=session,
        )
        self.log.debug("Getting dagrun for dag %s", self.dag_id)
        dr: DagRun = _get_or_create_dagrun(
            dag=self,
            start_date=execution_date,
            execution_date=execution_date,
            run_id=DagRun.generate_run_id(DagRunType.MANUAL, execution_date),
            session=session,
            conf=run_conf,
        )
        tasks = self.task_dict
        self.log.debug("starting dagrun")
        
        
        
        while dr.state == State.RUNNING:
            schedulable_tis, _ = dr.update_state(session=session)
            for ti in schedulable_tis:
                add_logger_if_needed(ti)
                ti.task = tasks[ti.task_id]
                _run_task(ti, session=session)
        if conn_file_path or variable_file_path:
            
            secrets_backend_list.pop(0)
    @provide_session
    def create_dagrun(