Source code for brewtils.errors

# -*- coding: utf-8 -*-

import json
import logging

from six import string_types


[docs]class BrewtilsException(Exception): """Base exception""" pass
# Error Logging Control
[docs]class SuppressStacktrace(Exception): """Mixin that will suppress stacktrace logging""" _bg_suppress_stacktrace = True
[docs]class ErrorLogLevelCritical(Exception): """Mixin to log an exception at the CRITICAL level""" _bg_error_log_level = logging.CRITICAL
[docs]class ErrorLogLevelError(Exception): """Mixin to log an exception at the ERROR level""" _bg_error_log_level = logging.ERROR
[docs]class ErrorLogLevelWarning(Exception): """Mixin to log an exception at the WARNING level""" _bg_error_log_level = logging.WARNING
[docs]class ErrorLogLevelInfo(Exception): """Mixin to log an exception at the INFO level""" _bg_error_log_level = logging.INFO
[docs]class ErrorLogLevelDebug(Exception): """Mixin to log an exception at the DEBUG level""" _bg_error_log_level = logging.DEBUG
# Models
[docs]class ModelError(BrewtilsException): """Base exception for model errors""" pass
[docs]class ModelValidationError(ModelError): """Invalid model""" pass
[docs]class RequestStatusTransitionError(ModelValidationError): """A status update was an invalid transition""" pass
# Plugins
[docs]class PluginError(BrewtilsException): """Generic error class""" pass
[docs]class PluginValidationError(PluginError): """Plugin could not be validated successfully""" pass
[docs]class PluginParamError(PluginError): """Error used when plugins have illegal parameters""" pass
# Requests
[docs]class RequestProcessException(BrewtilsException): """Base for exceptions that occur during request processing""" pass
[docs]class AckAndContinueException(RequestProcessException): pass
[docs]class NoAckAndDieException(RequestProcessException): pass
[docs]class AckAndDieException(RequestProcessException): pass
[docs]class DiscardMessageException(RequestProcessException): """Raising an instance will result in a message not being requeued""" pass
[docs]class RepublishRequestException(RequestProcessException): """Republish to the end of the message queue :param request: The Request to republish :param headers: A dictionary of headers to be used by `brewtils.request_consumer.RequestConsumer` :type request: :py:class:`brewtils.models.Request` """ def __init__(self, request, headers): self.request = request self.headers = headers
[docs]class RequestProcessingError(AckAndContinueException): pass
[docs]class RequestPublishException(BrewtilsException): """Error while publishing request""" pass
# Rest / Client errors
[docs]class RestError(BrewtilsException): """Base exception for REST errors""" pass
[docs]class RestClientError(RestError): """Wrapper for all 4XX errors""" pass
[docs]class RestServerError(RestError): """Wrapper for all 5XX errors""" pass
[docs]class RestConnectionError(RestServerError): """Error indicating a connection error while performing a request""" pass
[docs]class FetchError(RestError): """Error Indicating a server Error occurred performing a GET""" pass
[docs]class ValidationError(RestClientError): """Error Indicating a client (400) Error occurred performing a POST/PUT""" pass
[docs]class SaveError(RestServerError): """Error Indicating a server Error occurred performing a POST/PUT""" pass
[docs]class DeleteError(RestServerError): """Error Indicating a server Error occurred performing a DELETE""" pass
[docs]class TimeoutExceededError(RestClientError): """Error indicating a timeout occurred waiting for a request to complete""" pass
[docs]class ConflictError(RestClientError): """Error indicating a 409 was raised on the server""" pass
[docs]class RequestFailedError(RestError): """Request returned with a 200, but the status was ERROR""" def __init__(self, request): self.request = request
[docs]class NotFoundError(RestClientError): """Error Indicating a 404 was raised on the server""" pass
[docs]class RequestForbidden(RestClientError): """Error indicating a 403 was raised on the server""" pass
[docs]class AuthorizationRequired(RestClientError): """Error indicating a 401 was raised on the server""" pass
# Alias old names WaitExceededError = TimeoutExceededError ConnectionTimeoutError = TimeoutExceededError BrewmasterModelError = ModelError BrewmasterModelValidationError = ModelValidationError BrewmasterRestError = RestError BrewmasterRestClientError = RestClientError BrewmasterRestServerError = RestServerError BrewmasterConnectionError = RestConnectionError BrewmasterTimeoutError = ConnectionTimeoutError BrewmasterFetchError = FetchError BrewmasterValidationError = ValidationError BrewmasterSaveError = SaveError BrewmasterDeleteError = DeleteError BGConflictError = ConflictError BGRequestFailedError = RequestFailedError BGNotFoundError = NotFoundError
[docs]def parse_exception_as_json(exc): """ Attempt to parse an Exception to a JSON string. If the exception has a single argument, no attributes, and the attribute can be converted to a valid JSON string, then that will be returned. Otherwise, a string version of the following form will be returned: { "message": "", "arguments": [], "attributes": {} } Where "message" is just str(exc), "arguments" is a list of all the arguments passed to the exception attempted to be converted to a valid JSON string, and "attributes" are the attributes of the exception class. If parsing fails at all, then a simple str() will be applied either the argument or attribute value. Note: On python version 2, errors with custom attributes do not list those attributes as arguments. Args: exc (Exception): The exception you would like to format as JSON. Raises: ValueError: If the exception passed in is not an Exception. Returns: A valid JSON string representing (the best we can) the exception. """ if not isinstance(exc, Exception): raise ValueError("Attempted to parse a non-exception as JSON.") json_args = [] valid_json = True for arg in exc.args: valid_json, json_arg = _jsonify_value(arg) json_args.append(json_arg) if ( len(json_args) == 1 and not exc.__dict__ and valid_json and isinstance(json_args[0], (list, dict)) ): return json.dumps(json_args[0]) return json.dumps( { "message": str(exc), "arguments": json_args, "attributes": _jsonify_value(exc.__dict__)[1], } )
def _jsonify_value(value): """Attempt to JSONify a value, returns success and then a string""" try: if isinstance(value, string_types): v = json.loads(value) if isinstance(v, string_types): return True, value else: return True, v else: json.dumps(value) return True, value except Exception: return False, str(value)