If we made [AirflowNotFound](https://github.com/apache/airflow/blob/master/airflow/exceptions.py#L43-L46) inherit from [connextion.NotFound](https://github.com/apache/airflow/blob/master/airflow/api_connexion/exceptions.py#L75-L83) this code could be a simpler [here](https://github.com/apache/airflow/blob/690e80b82b12ab1d7232d6f02ea6bddd1d71750c/airflow/api_connexion/endpoints/task_instance_endpoint.py#L290-L297).


_Originally posted by @ashb in https://github.com/apache/airflow/pull/11335#discussion_r511863077_
airflow/api_connexion/endpoints/task_instance_endpoint.py
data = clear_task_instance_form.load(body)
    except ValidationError as err:
        raise BadRequest(detail=str(err.messages))
    dag = current_app.dag_bag.get_dag(dag_id)
    if not dag:
        error_message = f"Dag id {dag_id} not found"
        raise NotFound(error_message)
    reset_dag_runs = data.pop('reset_dag_runs')
    dry_run = data.pop('dry_run')
    
    task_instances = dag.clear(dry_run=True, dag_bag=current_app.dag_bag, **data)
    if not dry_run:
        clear_task_instances(
            task_instances.all(), session, dag=dag, dag_run_state=State.RUNNING if reset_dag_runs else False
        )
    task_instances = task_instances.join(TI.dag_run).options(eagerload(TI.dag_run))
    return task_instance_reference_collection_schema.dump(
        TaskInstanceReferenceCollection(task_instances=task_instances.all())
    )
@security.requires_access(
    [
        (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAG),
        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
        (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_TASK_INSTANCE),
    ]
)
@provide_session
def post_set_task_instances_state(dag_id, session):
    """Set a state of task instances."""
    body = request.get_json()