Source code for axonius_api_client.exceptions

# -*- 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 ToolsError(AxonError): """Errors for tools."""
[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 SavedQueryTagsNotFoundError(SavedQueryNotFoundError): """Error when something is not found."""
[docs] def __init__(self, value: List[str], valid: List[str]): """Pass.""" self.value = value self.valid = valid value_txt = ", ".join(value) valid_txt = "\n" + "\n".join(valid) self.msg = f"Saved Query not found with tags: {value_txt}, valid tags:{valid_txt}" super(NotFoundError, self).__init__(self.msg)
[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 ApiAttributeExtraWarning(ApiWarning): """Pass."""
[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 = 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 RequestFormatError(RequestError): """Pass."""
[docs]class RequestFormatPathError(RequestFormatError): """Pass."""
[docs]class RequestFormatObjectError(RequestFormatError): """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 to enable" ) super().__init__(msg)