Repository: apache/airflow
File: providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py (Python)

Code before:

    @staticmethod
    def _base_url(instance_url: str, namespace: str) -> str:
        return os.path.join(instance_url, "v3", "namespaces", quote(namespace), "apps")

    def _cdap_request(
        self, url: str, method: str, body: list | dict | None = None, params: dict | None = None
    ) -> google.auth.transport.Response:
        headers: dict[str, str] = {"Content-Type": "application/json"}
        request = google.auth.transport.requests.Request()

        credentials = self.get_credentials()
        credentials.before_request(request=request, method=method, url=url, headers=headers)

        payload = json.dumps(body) if body else None

        response = request(method=method, url=url, headers=headers, body=payload, params=params)
        return response

    @staticmethod
    def _check_response_status_and_data(response, message: str) -> None:
        if not response:
            raise AirflowException(
                "Invalid / Empty response received. Please, check for possible root "
                "causes of this behavior either in DAG code or on Cloud DataFusion side"
            )
        if response.status == 404:
            raise AirflowNotFoundException(message)
        if response.status == 409:
            raise ConflictException("Conflict: Resource is still in use.")
        if response.status != 200:
            raise AirflowException(message)
        if response.data is None:
            raise AirflowException(
                "Empty response received. Please, check for possible root "
                "causes of this behavior either in DAG code or on Cloud DataFusion side"
            )

    def get_conn(self) -> Resource:
        """Retrieve connection to DataFusion."""
        if not self._conn:
            http_authorized = self._authorize()
            self._conn = build(
                "datafusion",
                self.api_version,
                http=http_authorized,
                cache_discovery=False,
            )
        return self._conn

    @GoogleBaseHook.fallback_to_default_project_id

Review comment (refactor):
I'd like to ask to refactor all usages of `AirflowException` in the code, as according to a [recent policy](https://github.com/astronomer/airflow/blob/d8b768b71509b6895744d4d8a32f06a5daeebb12/contributing-docs/05_pull_requests.rst#dont-raise-airflowexception-directly) we want to avoid further usage of this exception at all (more about that in the [dev list thread](https://lists.apache.org/thread/5rv4tz0oc27bgr4khx0on0jz8fpxvh55)).
You may use a simple `ValueError` exception instead.

Code after:
        return f"projects/{project_id}/locations/{location}/instances/{instance_name}"

    @staticmethod
    def _parent(project_id: str, location: str) -> str:
        return f"projects/{project_id}/locations/{location}"

    @staticmethod
    def _base_url(instance_url: str, namespace: str) -> str:
        return os.path.join(instance_url, "v3", "namespaces", quote(namespace), "apps")

    def _cdap_request(
        self, url: str, method: str, body: list | dict | None = None, params: dict | None = None
    ) -> google.auth.transport.Response:
        headers: dict[str, str] = {"Content-Type": "application/json"}
        request = google.auth.transport.requests.Request()

        credentials = self.get_credentials()
        credentials.before_request(request=request, method=method, url=url, headers=headers)

        payload = json.dumps(body) if body else None

        response = request(method=method, url=url, headers=headers, body=payload, params=params)
        return response

    @staticmethod
    def _check_response_status_and_data(response, message: str) -> None:
        if not response:
            raise RequestException(
                "Invalid / Empty response received. Please, check for possible root "
                "causes of this behavior either in DAG code or on Cloud DataFusion side"
            )
        if response.status == 404:
            raise HTTPError(message)
        if response.status == 409:
            raise ConflictException("Conflict: Resource is still in use.")
        if response.status != 200:
            raise RequestException(message)
        if response.data is None:
            raise ValueError(
                "Empty response received. Please, check for possible root "
                "causes of this behavior either in DAG code or on Cloud DataFusion side"
            )

    def get_conn(self) -> Resource:
        """Retrieve connection to DataFusion."""
        if not self._conn:
            http_authorized = self._authorize()
            self._conn = build(
                "datafusion",
                self.api_version,
                http=http_authorized,