/usr/lib/python2.7/dist-packages/taskflow/exceptions.py is in python-taskflow 0.1.3-0ubuntu3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | # -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import six
class TaskFlowException(Exception):
"""Base class for exceptions emitted from this library."""
pass
class ConnectionFailure(TaskFlowException):
"""Raised when some type of connection can not be opened or is lost."""
pass
class Duplicate(TaskFlowException):
"""Raised when a duplicate entry is found."""
pass
class StorageError(TaskFlowException):
"""Raised when logbook can not be read/saved/deleted."""
def __init__(self, message, cause=None):
super(StorageError, self).__init__(message)
self.cause = cause
class NotFound(TaskFlowException):
"""Raised when some entry in some object doesn't exist."""
pass
class AlreadyExists(TaskFlowException):
"""Raised when some entry in some object already exists."""
pass
class InvalidState(TaskFlowException):
"""Raised when a task/job/workflow is in an invalid state when an
operation is attempting to apply to said task/job/workflow.
"""
pass
class InvariantViolation(TaskFlowException):
"""Raised when flow invariant violation is attempted."""
pass
class UnclaimableJob(TaskFlowException):
"""Raised when a job can not be claimed."""
pass
class JobNotFound(TaskFlowException):
"""Raised when a job entry can not be found."""
pass
class MissingDependencies(InvariantViolation):
"""Raised when a entity has dependencies that can not be satisfied."""
message = ("%(who)s requires %(requirements)s but no other entity produces"
" said requirements")
def __init__(self, who, requirements):
message = self.message % {'who': who, 'requirements': requirements}
super(MissingDependencies, self).__init__(message)
self.missing_requirements = requirements
class DependencyFailure(TaskFlowException):
"""Raised when flow can't resolve dependency."""
pass
class EmptyFlow(TaskFlowException):
"""Raised when flow doesn't contain tasks."""
pass
class WrappedFailure(TaskFlowException):
"""Wraps one or several failures.
When exception cannot be re-raised (for example, because
the value and traceback is lost in serialization) or
there are several exceptions, we wrap corresponding Failure
objects into this exception class.
"""
def __init__(self, causes):
self._causes = []
for cause in causes:
if cause.check(type(self)) and cause.exception:
# NOTE(imelnikov): flatten wrapped failures.
self._causes.extend(cause.exception)
else:
self._causes.append(cause)
def __iter__(self):
"""Iterate over failures that caused the exception."""
return iter(self._causes)
def __len__(self):
"""Return number of wrapped failures."""
return len(self._causes)
def check(self, *exc_classes):
"""Check if any of exc_classes caused (part of) the failure.
Arguments of this method can be exception types or type names
(stings). If any of wrapped failures were caused by exception
of given type, the corresponding argument is returned. Else,
None is returned.
"""
if not exc_classes:
return None
for cause in self:
result = cause.check(*exc_classes)
if result is not None:
return result
return None
def __str__(self):
causes = [exception_message(cause) for cause in self._causes]
return 'WrappedFailure: %s' % causes
def exception_message(exc):
"""Return the string representation of exception."""
# NOTE(imelnikov): Dealing with non-ascii data in python is difficult:
# https://bugs.launchpad.net/taskflow/+bug/1275895
# https://bugs.launchpad.net/taskflow/+bug/1276053
try:
return six.text_type(exc)
except UnicodeError:
return str(exc)
|