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
try:
        data = set_task_instance_state_form.load(body)
    except ValidationError as err:
        raise BadRequest(detail=str(err.messages))
    error_message = f"Dag ID {dag_id} not found"
    try:
        dag = current_app.dag_bag.get_dag(dag_id)
        if not dag:
            raise NotFound(error_message)
    except SerializedDagNotFound:
        
        raise NotFound(error_message)
    task_id = data['task_id']
    task = dag.task_dict.get(task_id)
    if not task:
        error_message = f"Task ID {task_id} not found"
        raise NotFound(error_message)
    tis = dag.set_task_instance_state(
        task_id=task_id,
        execution_date=data["execution_date"],
        state=data["new_state"],
        upstream=data["include_upstream"],
        downstream=data["include_downstream"],
        future=data["include_future"],
        past=data["include_past"],
        commit=not data["dry_run"],
        session=session,
    )
    return task_instance_reference_collection_schema.dump(TaskInstanceReferenceCollection(task_instances=tis))