Module astrapy.exceptions.devops_api_exceptions

Expand source code
# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

import httpx


class DevOpsAPIException(ValueError):
    """
    An exception specific to issuing requests to the DevOps API.
    """

    def __init__(self, text: str | None = None):
        super().__init__(text or "")


@dataclass
class DevOpsAPIErrorDescriptor:
    """
    An object representing a single error returned from the DevOps API,
    typically with an error code and a text message.

    A single response from the Devops API may return zero, one or more of these.

    Attributes:
        id: a numeric code as found in the API "ID" item.
        message: the text found in the API "error" item.
        attributes: a dict with any further key-value pairs returned by the API.
    """

    id: int | None
    message: str | None
    attributes: dict[str, Any]

    def __init__(self, error_dict: dict[str, Any]) -> None:
        self.id = error_dict.get("ID")
        self.message = error_dict.get("message")
        self.attributes = {
            k: v for k, v in error_dict.items() if k not in {"ID", "message"}
        }


@dataclass
class DevOpsAPIHttpException(DevOpsAPIException, httpx.HTTPStatusError):
    """
    A request to the DevOps API resulted in an HTTP 4xx or 5xx response.

    Though the DevOps API seldom enriches such errors with a response text,
    this class acts as the DevOps counterpart to DataAPIHttpException
    to facilitate a symmetryc handling of errors at application lebel.

    Attributes:
        text: a text message about the exception.
        error_descriptors: a list of all DevOpsAPIErrorDescriptor objects
            found in the response.
    """

    text: str | None
    error_descriptors: list[DevOpsAPIErrorDescriptor]

    def __init__(
        self,
        text: str | None,
        *,
        httpx_error: httpx.HTTPStatusError,
        error_descriptors: list[DevOpsAPIErrorDescriptor],
    ) -> None:
        DevOpsAPIException.__init__(self, text)
        httpx.HTTPStatusError.__init__(
            self,
            message=str(httpx_error),
            request=httpx_error.request,
            response=httpx_error.response,
        )
        self.text = text
        self.httpx_error = httpx_error
        self.error_descriptors = error_descriptors

    def __str__(self) -> str:
        return self.text or str(self.httpx_error)

    @classmethod
    def from_httpx_error(
        cls,
        httpx_error: httpx.HTTPStatusError,
        **kwargs: Any,
    ) -> DevOpsAPIHttpException:
        """Parse a httpx status error into this exception."""

        raw_response: dict[str, Any]
        # the attempt to extract a response structure cannot afford failure.
        try:
            raw_response = httpx_error.response.json()
        except Exception:
            raw_response = {}
        error_descriptors = [
            DevOpsAPIErrorDescriptor(error_dict)
            for error_dict in raw_response.get("errors") or []
        ]
        if error_descriptors:
            text = f"{error_descriptors[0].message}. {str(httpx_error)}"
        else:
            text = str(httpx_error)

        return cls(
            text=text,
            httpx_error=httpx_error,
            error_descriptors=error_descriptors,
            **kwargs,
        )


@dataclass
class DevOpsAPITimeoutException(DevOpsAPIException):
    """
    A DevOps API operation timed out.

    Attributes:
        text: a textual description of the error
        timeout_type: this denotes the phase of the HTTP request when the event
            occurred ("connect", "read", "write", "pool") or "generic" if there is
            not a specific request associated to the exception.
        endpoint: if the timeout is tied to a specific request, this is the
            URL that the request was targeting.
        raw_payload:  if the timeout is tied to a specific request, this is the
            associated payload (as a string).
    """

    text: str
    timeout_type: str
    endpoint: str | None
    raw_payload: str | None

    def __init__(
        self,
        text: str,
        *,
        timeout_type: str,
        endpoint: str | None,
        raw_payload: str | None,
    ) -> None:
        super().__init__(text)
        self.text = text
        self.timeout_type = timeout_type
        self.endpoint = endpoint
        self.raw_payload = raw_payload


@dataclass
class UnexpectedDevOpsAPIResponseException(DevOpsAPIException):
    """
    The DevOps API response is malformed in that it does not have
    expected field(s), or they are of the wrong type.

    Attributes:
        text: a text message about the exception.
        raw_response: the response returned by the API in the form of a dict.
    """

    text: str
    raw_response: dict[str, Any] | None

    def __init__(
        self,
        text: str,
        raw_response: dict[str, Any] | None,
    ) -> None:
        super().__init__(text)
        self.text = text
        self.raw_response = raw_response


class DevOpsAPIResponseException(DevOpsAPIException):
    """
    A request to the DevOps API returned with a non-success return code
    and one of more errors in the HTTP response.

    Attributes:
        text: a text message about the exception.
        command: the raw payload that was sent to the DevOps API.
        error_descriptors: a list of all DevOpsAPIErrorDescriptor objects
            returned by the API in the response.
    """

    text: str | None
    command: dict[str, Any] | None
    error_descriptors: list[DevOpsAPIErrorDescriptor]

    def __init__(
        self,
        text: str | None = None,
        *,
        command: dict[str, Any] | None = None,
        error_descriptors: list[DevOpsAPIErrorDescriptor] = [],
    ) -> None:
        super().__init__(text or self.__class__.__name__)
        self.text = text
        self.command = command
        self.error_descriptors = error_descriptors

    @staticmethod
    def from_response(
        command: dict[str, Any] | None,
        raw_response: dict[str, Any],
    ) -> DevOpsAPIResponseException:
        """Parse a raw response from the API into this exception."""

        error_descriptors = [
            DevOpsAPIErrorDescriptor(error_dict)
            for error_dict in raw_response.get("errors") or []
        ]
        if error_descriptors:
            _text = error_descriptors[0].message
        else:
            _text = None
        return DevOpsAPIResponseException(
            text=_text, command=command, error_descriptors=error_descriptors
        )

Classes

class DevOpsAPIErrorDescriptor (error_dict: dict[str, Any])

An object representing a single error returned from the DevOps API, typically with an error code and a text message.

A single response from the Devops API may return zero, one or more of these.

Attributes

id
a numeric code as found in the API "ID" item.
message
the text found in the API "error" item.
attributes
a dict with any further key-value pairs returned by the API.
Expand source code
@dataclass
class DevOpsAPIErrorDescriptor:
    """
    An object representing a single error returned from the DevOps API,
    typically with an error code and a text message.

    A single response from the Devops API may return zero, one or more of these.

    Attributes:
        id: a numeric code as found in the API "ID" item.
        message: the text found in the API "error" item.
        attributes: a dict with any further key-value pairs returned by the API.
    """

    id: int | None
    message: str | None
    attributes: dict[str, Any]

    def __init__(self, error_dict: dict[str, Any]) -> None:
        self.id = error_dict.get("ID")
        self.message = error_dict.get("message")
        self.attributes = {
            k: v for k, v in error_dict.items() if k not in {"ID", "message"}
        }

Class variables

var attributes : dict[str, typing.Any]
var id : int | None
var message : str | None
class DevOpsAPIException (text: str | None = None)

An exception specific to issuing requests to the DevOps API.

Expand source code
class DevOpsAPIException(ValueError):
    """
    An exception specific to issuing requests to the DevOps API.
    """

    def __init__(self, text: str | None = None):
        super().__init__(text or "")

Ancestors

  • builtins.ValueError
  • builtins.Exception
  • builtins.BaseException

Subclasses

class DevOpsAPIHttpException (text: str | None, *, httpx_error: httpx.HTTPStatusError, error_descriptors: list[DevOpsAPIErrorDescriptor])

A request to the DevOps API resulted in an HTTP 4xx or 5xx response.

Though the DevOps API seldom enriches such errors with a response text, this class acts as the DevOps counterpart to DataAPIHttpException to facilitate a symmetryc handling of errors at application lebel.

Attributes

text
a text message about the exception.
error_descriptors
a list of all DevOpsAPIErrorDescriptor objects found in the response.
Expand source code
@dataclass
class DevOpsAPIHttpException(DevOpsAPIException, httpx.HTTPStatusError):
    """
    A request to the DevOps API resulted in an HTTP 4xx or 5xx response.

    Though the DevOps API seldom enriches such errors with a response text,
    this class acts as the DevOps counterpart to DataAPIHttpException
    to facilitate a symmetryc handling of errors at application lebel.

    Attributes:
        text: a text message about the exception.
        error_descriptors: a list of all DevOpsAPIErrorDescriptor objects
            found in the response.
    """

    text: str | None
    error_descriptors: list[DevOpsAPIErrorDescriptor]

    def __init__(
        self,
        text: str | None,
        *,
        httpx_error: httpx.HTTPStatusError,
        error_descriptors: list[DevOpsAPIErrorDescriptor],
    ) -> None:
        DevOpsAPIException.__init__(self, text)
        httpx.HTTPStatusError.__init__(
            self,
            message=str(httpx_error),
            request=httpx_error.request,
            response=httpx_error.response,
        )
        self.text = text
        self.httpx_error = httpx_error
        self.error_descriptors = error_descriptors

    def __str__(self) -> str:
        return self.text or str(self.httpx_error)

    @classmethod
    def from_httpx_error(
        cls,
        httpx_error: httpx.HTTPStatusError,
        **kwargs: Any,
    ) -> DevOpsAPIHttpException:
        """Parse a httpx status error into this exception."""

        raw_response: dict[str, Any]
        # the attempt to extract a response structure cannot afford failure.
        try:
            raw_response = httpx_error.response.json()
        except Exception:
            raw_response = {}
        error_descriptors = [
            DevOpsAPIErrorDescriptor(error_dict)
            for error_dict in raw_response.get("errors") or []
        ]
        if error_descriptors:
            text = f"{error_descriptors[0].message}. {str(httpx_error)}"
        else:
            text = str(httpx_error)

        return cls(
            text=text,
            httpx_error=httpx_error,
            error_descriptors=error_descriptors,
            **kwargs,
        )

Ancestors

  • DevOpsAPIException
  • builtins.ValueError
  • httpx.HTTPStatusError
  • httpx.HTTPError
  • builtins.Exception
  • builtins.BaseException

Class variables

var error_descriptors : list[DevOpsAPIErrorDescriptor]
var text : str | None

Static methods

def from_httpx_error(httpx_error: httpx.HTTPStatusError, **kwargs: Any) ‑> DevOpsAPIHttpException

Parse a httpx status error into this exception.

Expand source code
@classmethod
def from_httpx_error(
    cls,
    httpx_error: httpx.HTTPStatusError,
    **kwargs: Any,
) -> DevOpsAPIHttpException:
    """Parse a httpx status error into this exception."""

    raw_response: dict[str, Any]
    # the attempt to extract a response structure cannot afford failure.
    try:
        raw_response = httpx_error.response.json()
    except Exception:
        raw_response = {}
    error_descriptors = [
        DevOpsAPIErrorDescriptor(error_dict)
        for error_dict in raw_response.get("errors") or []
    ]
    if error_descriptors:
        text = f"{error_descriptors[0].message}. {str(httpx_error)}"
    else:
        text = str(httpx_error)

    return cls(
        text=text,
        httpx_error=httpx_error,
        error_descriptors=error_descriptors,
        **kwargs,
    )
class DevOpsAPIResponseException (text: str | None = None, *, command: dict[str, Any] | None = None, error_descriptors: list[DevOpsAPIErrorDescriptor] = [])

A request to the DevOps API returned with a non-success return code and one of more errors in the HTTP response.

Attributes

text
a text message about the exception.
command
the raw payload that was sent to the DevOps API.
error_descriptors
a list of all DevOpsAPIErrorDescriptor objects returned by the API in the response.
Expand source code
class DevOpsAPIResponseException(DevOpsAPIException):
    """
    A request to the DevOps API returned with a non-success return code
    and one of more errors in the HTTP response.

    Attributes:
        text: a text message about the exception.
        command: the raw payload that was sent to the DevOps API.
        error_descriptors: a list of all DevOpsAPIErrorDescriptor objects
            returned by the API in the response.
    """

    text: str | None
    command: dict[str, Any] | None
    error_descriptors: list[DevOpsAPIErrorDescriptor]

    def __init__(
        self,
        text: str | None = None,
        *,
        command: dict[str, Any] | None = None,
        error_descriptors: list[DevOpsAPIErrorDescriptor] = [],
    ) -> None:
        super().__init__(text or self.__class__.__name__)
        self.text = text
        self.command = command
        self.error_descriptors = error_descriptors

    @staticmethod
    def from_response(
        command: dict[str, Any] | None,
        raw_response: dict[str, Any],
    ) -> DevOpsAPIResponseException:
        """Parse a raw response from the API into this exception."""

        error_descriptors = [
            DevOpsAPIErrorDescriptor(error_dict)
            for error_dict in raw_response.get("errors") or []
        ]
        if error_descriptors:
            _text = error_descriptors[0].message
        else:
            _text = None
        return DevOpsAPIResponseException(
            text=_text, command=command, error_descriptors=error_descriptors
        )

Ancestors

Class variables

var command : dict[str, typing.Any] | None
var error_descriptors : list[DevOpsAPIErrorDescriptor]
var text : str | None

Static methods

def from_response(command: dict[str, Any] | None, raw_response: dict[str, Any]) ‑> DevOpsAPIResponseException

Parse a raw response from the API into this exception.

Expand source code
@staticmethod
def from_response(
    command: dict[str, Any] | None,
    raw_response: dict[str, Any],
) -> DevOpsAPIResponseException:
    """Parse a raw response from the API into this exception."""

    error_descriptors = [
        DevOpsAPIErrorDescriptor(error_dict)
        for error_dict in raw_response.get("errors") or []
    ]
    if error_descriptors:
        _text = error_descriptors[0].message
    else:
        _text = None
    return DevOpsAPIResponseException(
        text=_text, command=command, error_descriptors=error_descriptors
    )
class DevOpsAPITimeoutException (text: str, *, timeout_type: str, endpoint: str | None, raw_payload: str | None)

A DevOps API operation timed out.

Attributes

text
a textual description of the error
timeout_type
this denotes the phase of the HTTP request when the event occurred ("connect", "read", "write", "pool") or "generic" if there is not a specific request associated to the exception.
endpoint
if the timeout is tied to a specific request, this is the URL that the request was targeting.
raw_payload
if the timeout is tied to a specific request, this is the associated payload (as a string).
Expand source code
@dataclass
class DevOpsAPITimeoutException(DevOpsAPIException):
    """
    A DevOps API operation timed out.

    Attributes:
        text: a textual description of the error
        timeout_type: this denotes the phase of the HTTP request when the event
            occurred ("connect", "read", "write", "pool") or "generic" if there is
            not a specific request associated to the exception.
        endpoint: if the timeout is tied to a specific request, this is the
            URL that the request was targeting.
        raw_payload:  if the timeout is tied to a specific request, this is the
            associated payload (as a string).
    """

    text: str
    timeout_type: str
    endpoint: str | None
    raw_payload: str | None

    def __init__(
        self,
        text: str,
        *,
        timeout_type: str,
        endpoint: str | None,
        raw_payload: str | None,
    ) -> None:
        super().__init__(text)
        self.text = text
        self.timeout_type = timeout_type
        self.endpoint = endpoint
        self.raw_payload = raw_payload

Ancestors

Class variables

var endpoint : str | None
var raw_payload : str | None
var text : str
var timeout_type : str
class UnexpectedDevOpsAPIResponseException (text: str, raw_response: dict[str, Any] | None)

The DevOps API response is malformed in that it does not have expected field(s), or they are of the wrong type.

Attributes

text
a text message about the exception.
raw_response
the response returned by the API in the form of a dict.
Expand source code
@dataclass
class UnexpectedDevOpsAPIResponseException(DevOpsAPIException):
    """
    The DevOps API response is malformed in that it does not have
    expected field(s), or they are of the wrong type.

    Attributes:
        text: a text message about the exception.
        raw_response: the response returned by the API in the form of a dict.
    """

    text: str
    raw_response: dict[str, Any] | None

    def __init__(
        self,
        text: str,
        raw_response: dict[str, Any] | None,
    ) -> None:
        super().__init__(text)
        self.text = text
        self.raw_response = raw_response

Ancestors

Class variables

var raw_response : dict[str, typing.Any] | None
var text : str