# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2008-2009, SnapLogic, Inc. All rights reserved.
#
# See http://www.snaplogic.org for more information about
# the SnapLogic project.
#
# This program is free software, distributed under the terms of
# the GNU General Public License Version 2. See the LEGAL file
# at the top of the source tree.
#
# "SnapLogic" is a trademark of SnapLogic, Inc.
#
#
# $
# $Id: snap_exceptions.py 6844 2009-03-18 01:05:10Z jbrendel $
"""
Module for exception classes
"""
import sys
import traceback
import inspect
class SnapException(Exception):
"""
The SnapLogic exception base class.
All other SnapLogic exceptions are derived from this class, and are
prefixed with Snap.
This is the base class of all SnapLogic exceptions. It inherits from the
Python built-in Exception class, and follow its conventions, such as using
descriptive exception name, passing parameters for additional information,
etc. This base class additionally facilitates stacking up (or chaining)
exceptions together.
Chaining of exceptions can be accomplished in two ways. Firstly:
1. Take an existing exception 'a'.
2. Create a new exception 'b' (maybe you just cought 'a').
3. Write: b.previous = a
Note that this will work no matter if 'b' is a native exception or a
SnapException.
The second method of chaining exceptions is via the 'chain()' classmethod.
So, for example:
try:
... some error ...
except Exception, e:
raise SnapException.chain(SnapIOException("foo"), e)
This second method is very useful in loops.
"""
def __str__(self):
"""Return a string representation of the exception."""
return self._message + repr(self.args)
def __init__(self, mesg, *param):
"""
Initialize a new exception object.
@param mesg: This is a message that might appear in a log file.
@type mesg: string
@param param: List of parameters, that may be supplied to the
base exception class.
@type param: list
"""
Exception.__init__(self, *param)
self._tb_list = traceback.format_list(traceback.extract_stack())
self._message = mesg
self.previous = None # Use 'append' to chain exceptions.
self.is_logged = False # This is set by the logging module when the
# exception has been logged already, so that
# we can check and prevent it from being logged
# again.
def append(self, ex):
"""
Append an exception to the current exception.
This function tests whether the specified exception is a SnapException.
If not, a SnapNativeError exception is created and relevant information
of the native exception is stored in it.
The specified (or created) SnapException is then specified as the
'previous' exception of the current exception object, allowing for
chaining.
To traverse the chain, use the 'previous' property.
@param ex: An exception that we want to chain to this exception.
@type ex: Exception.
"""
self.previous = SnapException.snapify_exception(ex)
def set_message(self, mesg):
"""
Set the user-readable message for the exception.
The message is also set when the Exception is created. However,
this function allows us to change the message at a later time.
@param mesg: This is a message that might appear in a log file.
@type mesg: string
"""
self._message = mesg
def no_traceback(self):
"""Remove the traceback that was attached to the exception."""
self._tb_list = None
def get_all(self, print_tb=False, is_first=True):
"""
Return a string representing all the exceptions, even appended ones.
The traceback can also be included, if so desired.
@param print_tb: Flag indicating whether the traceback should
be printed as well.
@type print_tb: bool
"""
tmp_str = ""
#if print_tb and self._traceback: # and is_first:
if print_tb and self._tb_list: # and is_first:
tmp_str = str(self) + "\n"
# Is there a better way to get our current file name? The last couple of
# items in the stack trace of any of our exceptions take place right here
# in this file. So we can take our current file name from the last item
# in the stack trace.
snap_exceptions_source_file = __file__
if snap_exceptions_source_file.endswith(".pyc"):
# After it has been executed once, __file__ gets a pyc exception.
# So, if that's the case we need to strip the 'c' at the end, because
# that's not how a stack trace is printed.
snap_exceptions_source_file = snap_exceptions_source_file[:len(snap_exceptions_source_file)-1]
for f in self._tb_list:
# Ugly hack alert: We got the stack trace by just taking a
# copy of our current stack when we created the first exception.
# The problem is that this also contains a few frames and
# instructions in our normal exception class code. We don't
# want to display those, so we strip it off by looking for
# any frames and lines that come from this source file here.
if f.find(snap_exceptions_source_file) == -1:
tmp_str += f
prev = self.previous
if prev is not None and hasattr(prev, 'previous'):
if hasattr(prev, 'get_all'):
tmp_str += prev.get_all(print_tb, False)
return tmp_str
@classmethod
def snapify_exception(cls, ex):
"""
Check whether an exception is a SnapException and 'snapify' it if necessary.
A utility method, which checks if the passed-in exception is a SnapException.
If so, it returns this exception. If it is not a SnapException, it creates
one and transfers relevant information about the exception to the newly
created SnapException, which is then returned.
@param ex: An exception for checking and snapification or None
@type ex: Exception
@return: SnapException or None
"""
if ex:
if not isinstance(ex, SnapException):
tb = traceback.format_tb(sys.exc_info()[2])
ex = SnapNativeError(str(ex))
ex._tb_list = tb
return ex
@classmethod
def chain(cls, exp1, exp2):
"""
Chain exceptions together.
This classmethod is used to chain an old exception (which may have a
chain in itself dangling off of it) to a new one. Example:
ex_chain = None
for ... some loop ...:
try:
some error
except e:
ex_chain = SnapExceptions.chain(ex_chain, e)
As you can see, this comes in very handy in loops. If ex_chain is None (as it is
in the beginning) it will be handled correctly by the code.
Under other (non-loop) circumstances, just use the 'previous' property directly.
Please note that even non-Snap exceptions can be chained together like this. In the
case of non-snap exceptions, the same rules as for append() apply: A SnapNativeError
is created on its behalf, and the relevant information of the native exception is
stored in it.
@param exp1: The old exception (chain).
@type exp1: Exception
@param exp2: The new exception.
@type exp2: Exception.
"""
exp2 = SnapException.snapify_exception(exp2)
if exp1:
if exp2:
exp2.append(exp1)
return exp2
#
# SnapLogic exceptions
#
# General errors
class SnapGeneralError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "General error: ", *param)
class SnapNativeError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "Native error: ", *param)
class SnapObjNotFoundError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "Object not found: ", *param)
class SnapObjExistsError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "Object already exists: ", *param)
class SnapObjPermissionError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "Object permission error: ", *param)
class SnapObjTypeError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "Object type error: ", *param)
class SnapFormatError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "Format error: ", *param)
class SnapValueError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "Value error: ", *param)
class SnapIOError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "IO error: ", *param)
class SnapEOFError(SnapIOError):
def __init__(self, *param):
SnapException.__init__(self, "End of file error: ", *param)
class SnapHttpErrorCode(SnapGeneralError):
def __init__(self, code, mesg):
SnapException.__init__(self, "HttpErrorCode: ", code, mesg)
self.http_code = code
self.err_message = mesg
class SnapNotAllowedError(SnapGeneralError):
def __init__(self, *param):
SnapException.__init__(self, "Not allowed error: ", *param)
class SnapInvalidURIError(SnapValueError):
def __init__(self, *param):
SnapException.__init__(self, "Invalid URI: ", *param)
# Component errors
class SnapComponentError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "Component error: ", *param)
class SnapComponentConfigError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "Component config error: ", *param)
# Repository errors
class SnapRepError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "Repository error: ", *param)
class SnapRepInitError(SnapRepError):
def __init__(self, *param):
SnapException.__init__(self, "Repository initialization error: ", *param)
class SnapRepInternalError(SnapRepError):
def __init__(self, *param):
SnapException.__init__(self, "Repository internal error: ", *param)
class SnapRepUpgradeError(SnapRepError):
def __init__(self, *param):
SnapException.__init__(self, "Repository upgrade error: ", *param)
class SnapRepConnectionError(SnapRepError):
def __init__(self, *param):
SnapException.__init__(self, "Repository disconnected: ", *param)
class SnapRepNotFoundError(SnapRepError):
def __init__(self, *param):
SnapException.__init__(self, "Repository not found: ", *param)
class SnapRepInvalidPackageError(SnapRepError):
def __init__(self, *param):
SnapException.__init__(self, "Invalid repository export package: ", *param)
# Auth object errors
class SnapAuthObjError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "Auth object error: ", *param)
# Resource errors
class SnapResDefError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "Resource error: ", *param)
class SnapResDefGUIDError(SnapResDefError):
def __init__(self, *param):
SnapException.__init__(self, "Resource GUID error:", *param)
class SnapResDefGenIDError(SnapResDefError):
def __init__(self, *param):
SnapException.__init__(self, "Resource generation ID error:", *param)
# Snap Plugin Errors
class SnapPluginError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "Snap Plugin error: ", *param)
# SnapStream errors
class SnapStreamError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "SnapStream error: ", *param)
class SnapStreamNegotiationError(SnapStreamError):
def __init__(self, *param):
SnapException.__init__(self, "SnapStream content negotiation error: ", *param)
class SnapStreamConnectionError(SnapStreamError):
def __init__(self, *param):
SnapException.__init__(self, "SnapStream connection error: " *param)
class SnapStreamIOError(SnapStreamError):
def __init__(self, *param):
SnapException.__init__(self, "SnapStream I/O error: ", *param)
class SnapStreamModeError(SnapStreamError):
def __init__(self, *param):
SnapException.__init__(self, "SnapStream stream mode error: ", *param)
# Notification errors
class SnapNotificationError(SnapException):
def __init__(self, *param):
SnapException.__init__(self, "SnapNotification error: ", *param)
|