import json
[docs]class AmsException(Exception):
"""Base exception class for all Argo Messaging service related errors"""
[docs] def __init__(self, *args, **kwargs):
super(AmsException, self).__init__(*args, **kwargs)
[docs]class AmsServiceException(AmsException):
"""Exception for Argo Messaging Service API errors"""
[docs] def __init__(self, json, request):
errord = dict()
self.msg = "While trying the [{0}]: {1}".format(request, json['error']['message'])
errord.update(error=self.msg)
if json['error'].get('code'):
self.code = json['error']['code']
errord.update(status_code=self.code)
if json['error'].get('status'):
self.status = json['error']['status']
errord.update(status=self.status)
super(AmsServiceException, self).__init__(errord)
[docs]class AmsBalancerException(AmsServiceException):
"""Exception for load balancer Argo Messaging Service errors"""
[docs] def __init__(self, json, request):
super(AmsBalancerException, self).__init__(json, request)
[docs]class AmsTimeoutException(AmsServiceException):
"""Exception for timeouts errors
Timeouts can be generated by the Argo Messaging Service if message was
not acknownledged in desired time frame (ackDeadlineSeconds). Also, 408
timeouts can come from load balancer for partial requests that were not
completed in required time frame.
"""
[docs] def __init__(self, json, request):
super(AmsTimeoutException, self).__init__(json, request)
[docs]class AmsConnectionException(AmsException):
"""Exception for connection related problems catched from requests library"""
[docs] def __init__(self, exp, request):
self.msg = "While trying the [{0}]: {1}".format(request, repr(exp))
super(AmsConnectionException, self).__init__(self.msg)
[docs]class AmsMessageException(AmsException):
"""Exception that indicate problems with constructing message"""
[docs] def __init__(self, msg):
self.msg = msg
super(AmsMessageException, self).__init__(self.msg)