return dag_run, False
        elif not create_if_necessary:
            raise DagRunNotFound(
                f"DagRun for {dag.dag_id} with run_id or execution_date "
                f"of {exec_date_or_run_id!r} not found"
            )
    if execution_date is not None:
        dag_run_execution_date = execution_date
    else:
        dag_run_execution_date = pendulum.instance(timezone.utcnow())
    if create_if_necessary == "memory":
        dag_run = DagRun(
            dag.dag_id,
            run_id=exec_date_or_run_id,
            execution_date=dag_run_execution_date,
            data_interval=dag.timetable.infer_manual_data_interval(run_after=dag_run_execution_date),
        )
        return dag_run, True
    elif create_if_necessary == "db":
        dag_run = dag.create_dagrun(
            state=DagRunState.QUEUED,
            execution_date=dag_run_execution_date,
            run_id=_generate_temporary_run_id(),
            data_interval=dag.timetable.infer_manual_data_interval(run_after=dag_run_execution_date),
            session=session,
        )
        return dag_run, True
    raise ValueError(f"unknown create_if_necessary value: {create_if_necessary!r}")
@provide_session
def _get_ti(