errors

Api errors.

Classes:

Name	Description
ApiError	Exception raised when an API request fails.
NotSavedError	Exception raised when the object was not saved into database prior to the current operation.

Functions:

Name	Description
handle_api_validation_errors	Raise HTTPValidationError in case the call to API return such response.
retry_on_exception	Retry function on certain exceptions.

Attributes:

Name	Type	Description
logger

logger = logging.getLogger(__name__)

ApiError(message: str)

Exception raised when an API request fails.

Source code in src/xpdeep/utils/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

NotSavedError(message: str)

Exception raised when the object was not saved into database prior to the current operation.

Source code in src/xpdeep/utils/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

handle_api_validation_errors(api_call_result: T | HTTPValidationError | None) -> T

Raise HTTPValidationError in case the call to API return such response.

Source code in src/xpdeep/utils/errors.py

def handle_api_validation_errors[T](api_call_result: T | HTTPValidationError | None) -> T:
    """Raise HTTPValidationError in case the call to API return such response."""
    if isinstance(api_call_result, HTTPValidationError):
        message = f"A ValidationError occurred, make sure that API client is up to date : {api_call_result.detail}"
        raise ApiError(message)

    if api_call_result is None:
        message = "Something went wrong. Unexpected API response type."
        raise ApiError(message)

    return api_call_result

retry_on_exception(exception_types: tuple[type[Exception], ...], max_retries: int = 10) -> Callable[[F], F]

Retry function on certain exceptions.

For instance: httpx.RemoteProtocolError: peer closed connection without sending complete message body (incomplete chunked read) We should retry the function.

Source code in src/xpdeep/utils/errors.py

def retry_on_exception[F](
    exception_types: tuple[type[Exception], ...],
    max_retries: int = 10,
) -> Callable[[F], F]:
    """Retry function on certain exceptions.

    For instance:
    `httpx.RemoteProtocolError`: peer closed connection without sending complete message body (incomplete chunked read)
    We should retry the function.
    """

    def decorator(func: F) -> F:
        @functools.wraps(func)  # type: ignore[arg-type]
        def wrapper(*args: object, **kwargs: object) -> object:
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)  # type: ignore[operator]
                except exception_types as e:
                    if attempt >= max_retries:
                        logger.warning(f"Failed after {max_retries} attempts due to: {e}")  # noqa: G004
                        raise
            return None

        return cast(F, wrapper)

    return decorator