airflow/api/common/experimental/mark_tasks.py
def _set_dag_run_state(dag_id, execution_date, state, session=None):
    """
    Helper method that set dag run state in the DB.
    :param dag_id: dag_id of target dag run
    :param execution_date: the execution date from which to start looking
    :param state: target state
    :param session: database session
    """
    DR = DagRun
    dr = session.query(DR).filter(
        DR.dag_id == dag_id,
        DR.execution_date == execution_date
    ).one()
    dr.state = state
    if state == State.RUNNING:
        dr.start_date = timezone.utcnow()
        dr.end_date = None
    else:
        dr.end_date = timezone.utcnow()
    session.merge(dr)
def _set_dag_run_state(dag_id, execution_date, state, session=None):
    """
    Helper method that set dag run state in the DB.
    :param dag_id: dag_id of target dag run
    :param execution_date: the execution date from which to start looking
    :param state: target state
    :param session: database session
    """
    DR = DagRun
    dr = session.query(DR).filter(
        DR.dag_id == dag_id,
        DR.execution_date == execution_date
    ).one()
    dr.state = state
    if state == State.RUNNING:
        dr.start_date = timezone.utcnow()
        dr.end_date = None
    else:
        dr.end_date = timezone.utcnow()
    session.merge(dr)
Helper method that set dag run state in the DB.
    :param dag_id: dag_id of target dag run
    :param execution_date: the execution date from which to start looking
    :param state: target state
    :param session: database session