here.platform.exceptions

Source code for here.platform.exceptions

Copyright (C) 2020-2022 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""This module contains all exception classes defined by the HERE Python SDK."""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Protocol

if TYPE_CHECKING:
from here.platform.catalog import PublicationState

_supported_formats_link = (
"https://docs.here.com/data-sdk/docs/here-platform-layers#supported-formats"
)

[docs]
class ResponseProtocol(Protocol):
"""Protocol defining the subset of Response API needed for exceptions."""

@property
def status_code(self) -> int:
"""HTTP status code of the response (e.g., 200, 404, 500)."""
...

@property
def reason(self) -> str:
"""HTTP reason phrase associated with the status code (e.g., 'OK', 'Not Found')."""
...

@property
def headers(self) -> Mapping[str, str]:
"""HTTP response headers as a mapping of header names to values."""
...

@property
def content(self) -> bytes:
"""Raw response body content as bytes."""
...

@property
def url(self) -> str:
"""The URL that was requested."""
...

@property
def text(self) -> str:
"""Response body content decoded as text."""
...

[docs]
class PlatformException(Exception):
"""Base class for HERE platform exceptions."""

def init(self, resp: ResponseProtocol):
"""
Instantiate PlatformException.
:param resp: response detail will be stored in this param
"""
self.resp = resp

def str(self) -> str:
"""
Return the message to be raised for this exception.

:return: error message
"""
message = (
"An error occurred with HERE platform: Status {status} - Reason {reason}\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

return message

[docs]
class AuthenticationException(PlatformException):
"""
This AuthenticationException is raised either authentication or authorization on the
platform fails.
"""

def init(self, resp):
"""
Instantiate AuthenticationException .
:param resp: response detail will be stored in this param
"""

self.resp = resp

def str(self) -> str:
"""
Return the message to be raised for this exception.

:return: error message
"""
return (
"An error occurred during authentication or authorization with HERE platform: "
"Status {status} - Reason {reason}\n Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class NonceAlreadyUsedException(AuthenticationException):
"""
This NonceAlreadyUsedException is raised if the server detects that the same nonce value
for a given timestamp.

This nonce check is intended to protect against replay attacks, ensuring that the parameters
for the signature (including nonce and timestamp) remain unique. However, it is possible that
this unintentionally happens, in which case a retry is required.
"""

Nonce already consumed. See:

AAA_ERROR_CODE = 401207

def init(self, resp):
"""
Instantiate NonceAlreadyUsedException .
:param resp: response detail will be stored in this param
"""

self.resp = resp

def str(self) -> str:
"""
Return the message to be raised for this exception.

:return: error message
"""
return (
"NonceAlreadyUsedException: Status {status} - Reason {reason}\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class ResourceLimitExceededException(PlatformException):
"""
This ResourceLimitExceededException is raised in below scenario
1.The maximum number of catalogs or layers possible for this organization
have already been created.
2.Maximum storage capacity allocation has been reached for this layer type for this realm.
"""

def init(self, resp):
"""
Instantiate ResourceLimitExceededException .
:param resp: response detail will be stored in this param
"""

self.resp = resp

def str(self) -> str:
"""
Return the message to be raised for this exception.

:return: error message
"""
return (
"ResourceLimitExceededException: Status:{status} - Reason: {reason}\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class DigestMismatchException(PlatformException):
"""
This DigestMismatchException is raised when the server detects that the digest provided for
a piece of data doesn't match the data itself, such as with an incomplete upload.
"""

def init(self, resp):
"""
Instantiate DigestMismatchException.
:param resp: response detail will be stored in this param
"""

self.resp = resp

def str(self) -> str:
"""
Return the message to be raised for this exception.

:return: error message
"""
return (
"DigestMismatchException: Status:{status} - Reason: {reason}\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class ConfigException(Exception):
"""
This ConfigException is raised whenever there is any error related to
platform configuration.
"""

[docs]
class ServiceUnavailableException(PlatformException):
"""Exception raised for API HTTP response status code 408, 500, 502, 503, or 504.

This is a dedicated exception to be used with the backoff package, because
it requires a specific exception class.
The exception value will be the response object returned by :mod:requests
which provides access to all its attributes, eg. :attr:status_code,
:attr:reason and :attr:text, etc.
"""

def str(self):
"""Return a string from the HTTP response causing the exception.
The string simply lists the response status code, reason and text
content, separated with commas.

:return: error message
"""

return (
"ServiceUnavailableException: Status
{status} - Reason {reason}\n\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class TooManyRequestsException(PlatformException):
"""Exception raised for API HTTP response status code 429.

This is a dedicated exception to be used with the backoff package, because
it requires a specific exception class.
The exception value will be the response object returned by :mod:requests
which provides access to all its attributes, eg. :attr:status_code,
:attr:reason and :attr:text, etc.
"""

def str(self):
"""Return a string from the HTTP response causing the exception.

The string simply lists the response status code, reason and text
content, separated with commas.
:return: error message
"""

return (
"TooManyRequestsException: Status
{status} - Reason {reason}\n\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class SchemaException(Exception):
"""
This SchemaException is raised when something about a data schema is invalid.
"""

[docs]
class EncodeException(Exception):
"""
Raised when data cannot be encoded to a target representation (content type).
"""

[docs]
class UnsupportedContentTypeEncodeException(EncodeException):
"""
Raised when trying to encode a content type that is not supported
by the specified :class:Adapter and corresponding :class:Encoder.
"""

def init(self, content_type: str):
"""Initialize the exception."""
self.content_type = content_type

def str(self):
"""Return a string representation of the exception"""
return (
f"Encoding content type {self.content_type} is not supported by the adapter used. "
"You might want encode the content manually and pass raw bytes to the write "
"function along with the parameter encode=False to skip automatic encoding. "
f"For the summary of supported formats, please see: {_supported_formats_link}"
)

[docs]
class DecodeException(Exception):
"""
Raised when data cannot be decoded from a source representation (content type).
"""

[docs]
class UnsupportedContentTypeDecodeException(DecodeException):
"""
Raised when trying to decode a content type that is not supported
by the specified :class:Adapter and corresponding :class:Decoder.
"""

def init(self, content_type: str):
"""Initialize the exception."""
self.content_type = content_type

def str(self):
"""Return a string representation of the exception"""
return (
f"Decoding content type {self.content_type} is not supported by the adapter used. "
"You might want to pass parameter decode=False to the read function "
"to skip automatic decoding, get raw bytes, and decode the content manually. "
f"For the summary of supported formats, please see: {_supported_formats_link}"
)

[docs]
class CatalogConfigurationException(Exception):
"""
This CatalogConfigurationException is raised when the catalog configuration is invalid.
"""

[docs]
class LayerConfigurationException(Exception):
"""
This LayerConfigurationException is raised when the layer configuration is invalid.
"""

[docs]
class PublicationException(Exception):
"""
This PublicationException is raised in case the publication is not finalized
successfully.
"""

def init(self, publication_id: str, publication_state: "PublicationState"):
"""
Instantiate PublicationException .
:param publication_id: a str representing publication id.
:param publication_state: a enum class representing publication state
"""
self.publication_state = publication_state
self.publication_id = publication_id

super().init(
f"Publication id: {self.publication_id} finished with state: "
f"{self.publication_state.value}"
)

[docs]
class UnsupportedOperationException(Exception):
"""
This UnsupportedOperationException is raised in case unsupported layer types are
used.
"""

def init(self, layer_id: str):
"""
Instantiate UnsupportedOperationException .
:param layer_id: layer ID.
"""
self.layer_id = layer_id
super().init(
f"""Currently this operation is not supported for layer ID: {layer_id}."""
)

[docs]
class PayloadTooLargeException(PlatformException):
"""Exception raised for API HTTP response status code 513.

This is a dedicated exception to be used for interactive map layer.
This exception will be raised when response payload is larger than the
specified limits of the interactive map layer.
The exception value will be the response object returned by :mod:requests
which provides access to all its attributes, eg. :attr:status_code,
:attr:reason and :attr:text, etc.
"""

def str(self):
"""Return a string from the HTTP response causing the exception.

The string simply lists the response status code, reason and text
content, separated with commas.
:return: error message
"""

return (
"PayloadTooLargeException: Status
{status} - Reason {reason}\n\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class RequestEntityTooLargeException(PlatformException):
"""Exception raised for API HTTP response status code 413.

This is a dedicated exception to be used for interactive map layer.
This exception will be raised when request body is larger than the
specified limits of the interactive map layer.
The exception value will be the response object returned by :mod:requests
which provides access to all its attributes, eg. :attr:status_code,
:attr:reason and :attr:text, etc.
"""

def str(self):
"""Return a string from the HTTP response causing the exception.

The string simply lists the response status code, reason and text
content, separated with commas.
:return: error message
"""

return (
"RequestEntityTooLargeException: Status
{status} - Reason {reason}\n\n"
"Response: {body}".format(
status=self.resp.status_code, reason=self.resp.reason, body=self.resp.text
)
)

[docs]
class InequalReadsException(PlatformException):
"""Exception raised when the response content actual length differs from expected length.

This Exception is used at request method at base api. Useful for large data size when there
are chances that inequal reads events might occur.
"""

def str(self):
"""
Return the message to be raised for this exception.

:return: error message
"""
expected_length = self.resp.headers.get("Content-Length")
expected_length = int(expected_length)
actual_length = int(self.resp.raw.tell())

if expected_length > actual_length:
reason = f"inequal read ({actual_length} bytes read,
{expected_length - actual_length} more expected)"
else:
reason = f"inequal read ({actual_length} bytes read,
{actual_length - expected_length} more than expected)"

message = (
"InequalReadsException: Status {status} - Reason {reason} \n"
"Response: {body}".format(
status=self.resp.status_code, reason=reason, body=self.resp.text
)
)
return message