"""Exceptions and warnings."""
import typing as t
import requests
[docs]def get_exc_str(exc: t.Optional[Exception] = None) -> str:
"""Pass."""
if exc:
return f"Original exception {type(exc)}: {exc}"
return ""
[docs]class AxonWarning(Warning):
"""Base class for all warnings in this package."""
[docs]class ApiWarning(AxonWarning):
"""Warnings for API models."""
[docs]class GuiQueryWizardWarning(ApiWarning):
"""Pass."""
[docs]class UnknownFieldSchema(ApiWarning):
"""Warning for unknown field schema mappings."""
[docs]class AxonError(Exception):
"""Base class for all exceptions in this package."""
[docs] def __init__(self, msg: t.Union[str, t.List[t.Any]]) -> None:
"""Pass."""
if isinstance(msg, (list, tuple)):
msg = "\n".join([str(x) for x in msg])
super().__init__(msg)
[docs]class AxonTypeError(AxonError):
"""Pass."""
[docs] def __init__(
self,
attr: str,
value: t.Any,
expected: t.Any,
src: t.Any = None,
extra: t.Any = None,
) -> None:
"""Pass."""
self.src: t.Any = src
self.attr: str = attr
self.value: t.Any = value
self.expected: t.Any = expected
self.extra: t.Any = extra
err: str = f"Incorrect value type supplied to {attr!r}"
msgs: t.List[str] = [
err,
"",
f"Source: {src}",
f"Supplied value: {value!r}",
f"Supplied value type: {type(value)}",
f"Expected value type: {expected!r}",
"",
err,
]
if extra:
if isinstance(extra, (list, tuple)):
msgs += [str(x) for x in extra]
else:
msgs.append(str(extra))
super().__init__(msgs)
[docs]class ApiError(AxonError):
"""Errors for API models."""
[docs]class ConfirmNotTrue(AxonError):
"""Error for when confirm != True."""
[docs] def __init__(
self,
confirm: t.Any = False,
prompt: t.Any = False,
reason: t.Any = "",
src: t.Any = None,
extra: t.Any = None,
) -> None:
"""Pass."""
self.confirm: t.Any = confirm
self.reason: t.Any = reason
msgs: t.List[str] = [
f"Unable to {reason}",
f"confirm is {confirm} and prompt is {prompt}, confirm must be {True}",
]
if src is not None:
msgs += ["", "While in object:", f"{src!r}"]
if extra:
if isinstance(extra, (list, tuple)):
msgs += [str(x) for x in extra]
else:
msgs.append(str(extra))
super().__init__(msgs)
[docs]class NotAllowedError(AxonError):
"""Error when something not allowed."""
[docs]class AuthError(AxonError):
"""Errors for authentication models."""
[docs]class NotFoundError(ApiError):
"""Error when something is not found."""
[docs]class SavedQueryNotFoundError(NotFoundError):
"""Error when something is not found."""
[docs] def __init__(self, details: str, sqs: t.List[t.Union[dict, object]]) -> None:
"""Pass."""
from .parsers.tables import tablize_sqs
self.sqs = sqs
self.details = details
self.msg = f"Saved Query not found with {details}"
try:
barrier = "#" * 80
barrier = f"\n{barrier}\n"
details = barrier + barrier.join([x.str_details for x in sqs])
self.msg_table = [self.msg, "", details, "", self.msg]
except Exception:
self.msg_table = tablize_sqs(data=sqs, err=self.msg)
super().__init__(self.msg_table)
[docs]class AlreadyExists(ApiError):
"""Error when something exists with same name."""
[docs]class NotLoggedIn(AuthError):
"""Error when not logged in."""
[docs]class AlreadyLoggedIn(AuthError):
"""Error when already logged in."""
[docs]class ConnectError(AxonError):
"""Error in connect client."""
[docs]class HttpError(AxonError):
"""Errors for HTTP client."""
[docs]class ConfigError(ApiError):
"""Errors in a configuration."""
[docs]class ConfigInvalidValue(ConfigError):
"""Error when a supplied configuration has a bad type or is the wrong choice."""
[docs]class ConfigUnchanged(ConfigError):
"""Error when a supplied configuration is no different from the current configuration."""
[docs]class ConfigUnknown(ConfigError):
"""Error when an unknown configuration key is supplied."""
[docs]class ConfigRequired(ConfigError):
"""Error when a required configuration key is not supplied."""
[docs]class CnxError(ApiError):
"""Errors for connections."""
[docs]class CnxGoneError(CnxError):
"""Errors when a connection has gone away."""
[docs]class CnxUpdateError(CnxError):
"""Errors when updating a connections configuration."""
[docs]class CnxTestError(CnxError):
"""Errors when testing a connections configuration."""
[docs]class CnxAddError(CnxError):
"""Errors when adding a new connection."""
cnx_new: t.ClassVar[t.Optional[dict]] = None
[docs]class ResponseError(ApiError):
"""Errors when checking responses."""
[docs] def __init__(
self,
msg: t.Optional[str] = None,
response=None,
exc: t.Optional[Exception] = None,
) -> None:
"""Error in responses received from REST API.
Args:
response (:obj:`requests.Response`): response that originated the error
msg: error message to include in exception
exc: original exception that was thrown if any
"""
self.response: requests.Response = response
self.exc: Exception = exc
self.msg: str = msg
self.errmsg: str = self.build_errmsg(response=response, msg=msg, exc=exc)
super().__init__(self.errmsg)
[docs] @classmethod
def build_errmsg(
cls,
response,
msg: t.Optional[str] = None,
exc: t.Optional[Exception] = None,
) -> str:
"""Build an error message from a response.
Args:
response (:obj:`requests.Response`): response that originated the error
msg: error message to include in exception
exc: exception that was thrown if any
"""
from .tools import json_log
url = response.url
method = response.request.method
code = response.status_code
reason = response.reason
out_len = len(response.request.body or "")
in_len = len(response.text or "")
msg = msg or "Error in REST API response"
pre = [
msg,
get_exc_str(exc=exc),
f"URL: {url!r}, METHOD: {method}",
f"CODE: {code!r}, REASON: {reason!r}, BYTES OUT: {out_len}, BYTES IN: {in_len}",
]
middle = [
"Request Object:",
json_log(obj=response.request.body),
"Response Object:",
json_log(obj=response.text),
]
msgs = [*pre, "", *middle, "", *pre]
return "\n".join(msgs)
[docs]class InvalidCredentials(ResponseError):
"""Error when credentials are invalid."""
[docs]class ResponseNotOk(ResponseError):
"""Error if response has a status code that is an error and error_status is True."""
[docs]class JsonInvalidError(ResponseError):
"""Error when response has invalid JSON."""
[docs]class JsonError(ResponseError):
"""Error when JSON has key:error that is not empty or key:status=error."""
[docs]class WizardError(ApiError):
"""Errors in query wizards."""
[docs]class ApiAttributeError(ApiError):
"""Pass."""
[docs]class ApiAttributeTypeError(ApiAttributeError):
"""Pass."""
[docs]class ApiAttributeMissingError(ApiAttributeError):
"""Pass."""
[docs]class NoTriggerDefinedError(ApiError):
"""Pass."""
[docs]class StopFetch(ApiError):
"""Pass."""
[docs] def __init__(self, reason: str, state: dict) -> None:
"""Pass."""
self.reason = reason
self.state = state
super().__init__(reason)
[docs]class SchemaError(ApiError):
"""Pass."""
[docs] def __init__(self, obj, schema, exc, data) -> None:
"""Pass."""
from .tools import json_log, prettify_obj
self.schema = schema
self.exc = exc
self.obj = obj
self.data = data
errors = []
if hasattr(exc, "messages"):
errors = exc.messages
if isinstance(errors, dict) and "errors" in errors:
errors = errors["errors"]
errors = prettify_obj(errors)
pre = f"Schema Validation Error in {schema}"
self.errors = [
pre,
f"From object: {obj}",
get_exc_str(exc=exc),
"",
"While trying to load data:",
f"{json_log(data)}",
*errors,
"",
pre,
]
self.msg = "\n".join(self.errors)
super().__init__(self.msg)
[docs]class RequestError(ApiError):
"""Pass."""
[docs] def __init__(
self,
api_endpoint,
err: str,
details: t.Optional[t.List[str]] = None,
exc: t.Optional[Exception] = None,
) -> None:
"""Pass."""
self.api_endpoint = api_endpoint
self.err = err
self.details = details or []
self.exc = exc
self.errors = [
err,
get_exc_str(exc=exc),
f"While in {api_endpoint}",
"",
*details,
"",
err,
]
self.msg = "\n".join(self.errors)
super().__init__(self.msg)
[docs]class RequestMissingArgsError(RequestError):
"""Pass."""
[docs]class RequestObjectTypeError(RequestError):
"""Pass."""
[docs]class RequestLoadObjectError(RequestError):
"""Pass."""
[docs]class ResponseLoadObjectError(RequestError):
"""Pass."""
[docs]class FeatureNotEnabledError(ApiError):
"""Pass."""
[docs] def __init__(self, name: str, extra: t.Optional[str] = None) -> None:
"""Pass."""
msg = (
f"The {name} feature is not enabled on this instance, "
"please contact support@axonius.com to enable"
)
if extra is not None:
msg = f"{msg}:\n{extra}"
super().__init__(msg)
[docs]class RunnerError(ApiError):
"""Pass."""
[docs]class RunnerWarning(ApiWarning):
"""Pass."""
[docs]class GrabberError(ApiError):
"""Pass."""
[docs]class GrabberWarning(ApiWarning):
"""Pass."""
[docs]class FolderAlreadyExistsError(AlreadyExists):
"""Error when something exists with same name."""
[docs]class FolderNotFoundError(NotFoundError):
"""Error when something is not found."""
[docs] def __init__(self, msg: str, folder: t.Optional[object] = None) -> None:
"""Pass."""
self.folder: t.Optional[object] = folder
super().__init__(msg)
[docs]class SearchError(AxonError):
"""Pass."""
[docs]class SearchUnmatchedError(SearchError):
"""Pass."""
[docs]class SearchNoMatchesError(SearchError):
"""Pass."""
[docs]class SearchNoObjectsError(SearchError):
"""Pass."""
[docs]class DecodeError(ValueError, AxonError):
"""Raised when a byte string cannot be decoded to a valid ObjectId string.
Attributes:
value (bytes): The input byte string.
encoding_format (str): The encoding format used for decoding.
encoding_errors (str): The error handling method used for decoding.
"""
[docs] def __init__(
self,
value: t.Any,
encoding_format: t.Optional[str] = None,
encoding_errors: t.Optional[str] = None,
) -> None:
"""Initialize a new instance of the DecodeError exception.
Args:
value (t.Any):
The input value.
encoding_format (t.Optional[str], optional):
The encoding format used for decoding.
encoding_errors (t.Optional[str], optional):
The error handling method used for decoding.
"""
from .tools import trim_value_repr
self.value: bytes = value
self.encoding_format: t.Optional[str] = encoding_format
self.encoding_errors: t.Optional[str] = encoding_errors
super().__init__(
f"The input byte string {trim_value_repr(self.value)} cannot be decoded to a valid "
f"ObjectId string using encoding_format {self.encoding_format!r} "
f" and encoding_errors {self.encoding_errors!r}.",
)
[docs]class InvalidObjectIdError(ValueError, AxonError):
"""Raised when an input string is not a valid ObjectId.
Attributes:
value (str): The input string.
"""
[docs] def __init__(self, value: str) -> None:
"""Initialize a new instance of the InvalidObjectIdError exception.
Args:
value (str): The input string.
"""
from .tools import trim_value_repr
self.value: t.Any = value
super().__init__(
f"The input string {trim_value_repr(self.value)} is not a valid ObjectId.",
)
[docs]class InvalidTypeError(TypeError, AxonError):
"""Raised when an input value is not one of the allowed types.
Attributes:
value (t.Any): The input value.
allowed_types (t.Tuple[type]): The tuple of allowed types.
"""
[docs] def __init__(
self,
value: t.Any,
allowed_types: t.Optional[t.Tuple[type]] = (),
) -> None:
"""Initialize a new instance of the InvalidTypeError exception.
Args:
value (t.Any): The input value.
allowed_types (t.Optional[t.Tuple[type]], optional): The tuple of allowed types.
"""
self.value: t.Any = value
self.allowed_types: t.Optional[t.Tuple[type]] = allowed_types
from .tools import get_type_str
super().__init__(
f"The input value {self.value!r} type {get_type_str(self.value)} is not "
f"one of the allowed types: {get_type_str(allowed_types)}.",
)