# -*- coding: utf-8 -*-
"""Exceptions and warnings."""
from typing import List, Optional, Union
import requests
[docs]def get_exc_str(exc: 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]class ApiError(AxonError):
"""Errors for API models."""
[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: List[Union[dict, object]]):
"""Pass."""
from .parsers.tables import tablize_sqs
self.sqs = sqs
self.details = details
self.msg = f"Saved Query not found with {details}"
self.tablemsg = tablize_sqs(data=sqs, err=self.msg)
super().__init__(self.tablemsg)
[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."""
[docs]class ResponseError(ApiError):
"""Errors when checking responses."""
[docs] def __init__(self, msg: Optional[str] = None, response=None, exc: Optional[Exception] = 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: Optional[str] = None, exc: 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):
"""Pass."""
self.reason = reason
self.state = state
super().__init__(reason)
[docs]class SchemaError(ApiError):
"""Pass."""
[docs] def __init__(self, obj, schema, exc, data):
"""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: Optional[List[str]] = None,
exc: Optional[Exception] = 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):
"""Pass."""
msg = (
f"The {name} feature is not enabled on this instance, "
"please contact support@axonius.com to enable"
)
super().__init__(msg)