Module astrapy.exceptions.data_api_exceptions

Classes

class CursorException (text: str, *, cursor_state: str)

Expand source code

@dataclass
class CursorException(DataAPIException):
    """
    The cursor operation cannot be invoked if a cursor is not in its pristine
    state (i.e. is already being consumed or is exhausted altogether).

    Attributes:
        text: a text message about the exception.
        cursor_state: a string description of the current state
            of the cursor. See the documentation for Cursor.
    """

    text: str
    cursor_state: str

    def __init__(
        self,
        text: str,
        *,
        cursor_state: str,
    ) -> None:
        super().__init__(text)
        self.text = text
        self.cursor_state = cursor_state

The cursor operation cannot be invoked if a cursor is not in its pristine state (i.e. is already being consumed or is exhausted altogether).

Attributes

text

a text message about the exception.

cursor_state

a string description of the current state of the cursor. See the documentation for Cursor.

Ancestors

• DataAPIException
• builtins.Exception
• builtins.BaseException

Instance variables

var cursor_state : str

The type of the None singleton.

var text : str

The type of the None singleton.

class DataAPIException (*args, **kwargs)

Expand source code

class DataAPIException(Exception):
    """
    Any exception occurred while issuing requests to the Data API
    and specific to it, such as:
      - a collection is found not to exist when gettings its metadata,
      - the API return a response with an error,
    but not, for instance,
      - a network error while sending an HTTP request to the API.
    """

    pass

Any exception occurred while issuing requests to the Data API and specific to it, such as: - a collection is found not to exist when gettings its metadata, - the API return a response with an error, but not, for instance, - a network error while sending an HTTP request to the API.

Ancestors

• builtins.Exception
• builtins.BaseException

Subclasses

• CollectionDeleteManyException
• CollectionInsertManyException
• CollectionUpdateManyException
• TooManyDocumentsToCountException
• CursorException
• DataAPIHttpException
• DataAPIResponseException
• DataAPITimeoutException
• UnexpectedDataAPIResponseException
• TableInsertManyException
• TooManyRowsToCountException

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

Expand source code

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

    In most cases this comes with additional information: the purpose
    of this class is to present such information in a structured way,
    akin to what happens for the DataAPIResponseException, while
    still raising (a subclass of) `httpx.HTTPStatusError`.

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

    text: str | None
    error_descriptors: list[DataAPIErrorDescriptor]

    def __init__(
        self,
        text: str | None,
        *,
        httpx_error: httpx.HTTPStatusError,
        error_descriptors: list[DataAPIErrorDescriptor],
    ) -> None:
        DataAPIException.__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,
    ) -> DataAPIHttpException:
        """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() or {}
        except Exception:
            raw_response = {}
        error_descriptors = [
            DataAPIErrorDescriptor(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,
        )

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

In most cases this comes with additional information: the purpose of this class is to present such information in a structured way, akin to what happens for the DataAPIResponseException, while still raising (a subclass of) httpx.HTTPStatusError.

Attributes

text

a text message about the exception.

error_descriptors

a list of all DataAPIErrorDescriptor objects found in the response.

Ancestors

• DataAPIException
• httpx.HTTPStatusError
• httpx.HTTPError
• builtins.Exception
• builtins.BaseException

Static methods

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

Parse a httpx status error into this exception.

Instance variables

var error_descriptors : list[DataAPIErrorDescriptor]

The type of the None singleton.

var text : str | None

The type of the None singleton.

class DataAPIResponseException (text: str | None,
*,
command: dict[str, Any] | None,
raw_response: dict[str, Any],
error_descriptors: list[DataAPIErrorDescriptor],
warning_descriptors: list[DataAPIWarningDescriptor])

Expand source code

@dataclass
class DataAPIResponseException(DataAPIException):
    """
    The Data API returned an HTTP 200 ("success") response, which however
    reports API-specific error(s), possibly alongside partial successes.

    Attributes:
        text: a text message about the exception.
        command: the payload to the API that led to the response.
        raw_response: the full response from the API.
        error_descriptors: a list of DataAPIErrorDescriptor, one for each
            item in the API response's "errors" field.
        warning_descriptors: a list of DataAPIWarningDescriptor, one for each
            item in the API response's "warnings" field (if there are any).
    """

    text: str | None
    command: dict[str, Any] | None
    raw_response: dict[str, Any]
    error_descriptors: list[DataAPIErrorDescriptor]
    warning_descriptors: list[DataAPIWarningDescriptor]

    def __init__(
        self,
        text: str | None,
        *,
        command: dict[str, Any] | None,
        raw_response: dict[str, Any],
        error_descriptors: list[DataAPIErrorDescriptor],
        warning_descriptors: list[DataAPIWarningDescriptor],
    ) -> None:
        super().__init__(text)
        self.text = text
        self.command = command
        self.raw_response = raw_response
        self.error_descriptors = error_descriptors
        self.warning_descriptors = warning_descriptors

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

        error_descriptors = [
            DataAPIErrorDescriptor(error_dict)
            for error_dict in (raw_response or {}).get("errors") or []
        ]
        warning_descriptors = [
            DataAPIWarningDescriptor(error_dict)
            for error_dict in (raw_response or {}).get("warnings") or []
        ]

        if error_descriptors:
            summaries = [e_d.summary() for e_d in error_descriptors]
            if len(summaries) == 1:
                text = summaries[0]
            else:
                _j_summaries = "; ".join(
                    f"[{summ_i + 1}] {summ_s}"
                    for summ_i, summ_s in enumerate(summaries)
                )
                text = f"[{len(summaries)} errors collected] {_j_summaries}"
        else:
            text = ""

        return DataAPIResponseException(
            text,
            command=command,
            raw_response=raw_response,
            error_descriptors=error_descriptors,
            warning_descriptors=warning_descriptors,
            **kwargs,
        )

The Data API returned an HTTP 200 ("success") response, which however reports API-specific error(s), possibly alongside partial successes.

Attributes

text

a text message about the exception.

command

the payload to the API that led to the response.

raw_response

the full response from the API.

error_descriptors

a list of DataAPIErrorDescriptor, one for each item in the API response's "errors" field.

warning_descriptors

a list of DataAPIWarningDescriptor, one for each item in the API response's "warnings" field (if there are any).

Ancestors

• DataAPIException
• builtins.Exception
• builtins.BaseException

Static methods

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

Expand source code

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

    error_descriptors = [
        DataAPIErrorDescriptor(error_dict)
        for error_dict in (raw_response or {}).get("errors") or []
    ]
    warning_descriptors = [
        DataAPIWarningDescriptor(error_dict)
        for error_dict in (raw_response or {}).get("warnings") or []
    ]

    if error_descriptors:
        summaries = [e_d.summary() for e_d in error_descriptors]
        if len(summaries) == 1:
            text = summaries[0]
        else:
            _j_summaries = "; ".join(
                f"[{summ_i + 1}] {summ_s}"
                for summ_i, summ_s in enumerate(summaries)
            )
            text = f"[{len(summaries)} errors collected] {_j_summaries}"
    else:
        text = ""

    return DataAPIResponseException(
        text,
        command=command,
        raw_response=raw_response,
        error_descriptors=error_descriptors,
        warning_descriptors=warning_descriptors,
        **kwargs,
    )

Parse a raw response from the API into this exception.

Instance variables

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

The type of the None singleton.

var error_descriptors : list[DataAPIErrorDescriptor]

The type of the None singleton.

var raw_response : dict[str, typing.Any]

The type of the None singleton.

var text : str | None

The type of the None singleton.

var warning_descriptors : list[DataAPIWarningDescriptor]

The type of the None singleton.

class DataAPITimeoutException (text: str, *, timeout_type: str, endpoint: str | None, raw_payload: str | None)

Expand source code

@dataclass
class DataAPITimeoutException(DataAPIException):
    """
    A Data API operation timed out. This can be a request timeout occurring
    during a specific HTTP request, or can happen over the course of a method
    involving several requests in a row, such as a paginated find.

    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

A Data API operation timed out. This can be a request timeout occurring during a specific HTTP request, or can happen over the course of a method involving several requests in a row, such as a paginated find.

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).

Ancestors

• DataAPIException
• builtins.Exception
• builtins.BaseException

Instance variables

var endpoint : str | None

The type of the None singleton.

var raw_payload : str | None

The type of the None singleton.

var text : str

The type of the None singleton.

var timeout_type : str

The type of the None singleton.

class UnexpectedDataAPIResponseException (text: str, raw_response: dict[str, Any] | None)

Expand source code

@dataclass
class UnexpectedDataAPIResponseException(DataAPIException):
    """
    The Data 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

The Data 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.

Ancestors

• DataAPIException
• builtins.Exception
• builtins.BaseException

Instance variables

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

The type of the None singleton.

var text : str

The type of the None singleton.