# :coding: utf-8
# :copyright: Copyright (c) 2014 ftrack
from builtins import str
import sys
import traceback
import ftrack_api.entity.base
[docs]class Error(Exception):
'''ftrack specific error.'''
default_message = 'Unspecified error occurred.'
[docs] def __init__(self, message=None, details=None):
'''Initialise exception with *message*.
If *message* is None, the class 'default_message' will be used.
*details* should be a mapping of extra information that can be used in
the message and also to provide more context.
'''
if message is None:
message = self.default_message
self.message = message
self.details = details
if self.details is None:
self.details = {}
self.traceback = traceback.format_exc()
def __str__(self):
'''Return string representation.'''
keys = {}
for key, value in self.details.items():
if isinstance(value, str):
value = value.encode(sys.getfilesystemencoding())
keys[key] = value
return str(self.message.format(**keys))
[docs]class AuthenticationError(Error):
'''Raise when an authentication error occurs.'''
default_message = 'Authentication error.'
[docs]class ServerError(Error):
'''Raise when the server reports an error.'''
default_message = 'Server reported error processing request.'
[docs]class ServerCompatibilityError(ServerError):
'''Raise when server appears incompatible.'''
default_message = 'Server incompatible.'
[docs]class NotFoundError(Error):
'''Raise when something that should exist is not found.'''
default_message = 'Not found.'
[docs]class NotUniqueError(Error):
'''Raise when unique value required and duplicate detected.'''
default_message = 'Non-unique value detected.'
[docs]class IncorrectResultError(Error):
'''Raise when a result is incorrect.'''
default_message = 'Incorrect result detected.'
[docs]class NoResultFoundError(IncorrectResultError):
'''Raise when a result was expected but no result was found.'''
default_message = 'Expected result, but no result was found.'
[docs]class MultipleResultsFoundError(IncorrectResultError):
'''Raise when a single result expected, but multiple results found.'''
default_message = 'Expected single result, but received multiple results.'
[docs]class EntityTypeError(Error):
'''Raise when an entity type error occurs.'''
default_message = 'Entity type error.'
[docs]class UnrecognisedEntityTypeError(EntityTypeError):
'''Raise when an unrecognised entity type detected.'''
default_message = 'Entity type "{entity_type}" not recognised.'
[docs] def __init__(self, entity_type, **kw):
'''Initialise with *entity_type* that is unrecognised.'''
kw.setdefault('details', {}).update(dict(
entity_type=entity_type
))
super(UnrecognisedEntityTypeError, self).__init__(**kw)
[docs]class OperationError(Error):
'''Raise when an operation error occurs.'''
default_message = 'Operation error.'
[docs]class InvalidStateError(Error):
'''Raise when an invalid state detected.'''
default_message = 'Invalid state.'
[docs]class InvalidStateTransitionError(InvalidStateError):
'''Raise when an invalid state transition detected.'''
default_message = (
'Invalid transition from {current_state!r} to {target_state!r} state '
'for entity {entity!r}'
)
[docs] def __init__(self, current_state, target_state, entity, **kw):
'''Initialise error.'''
kw.setdefault('details', {}).update(dict(
current_state=current_state,
target_state=target_state,
entity=entity
))
super(InvalidStateTransitionError, self).__init__(**kw)
[docs]class AttributeError(Error):
'''Raise when an error related to an attribute occurs.'''
default_message = 'Attribute error.'
[docs]class ImmutableAttributeError(AttributeError):
'''Raise when modification of immutable attribute attempted.'''
default_message = (
'Cannot modify value of immutable {attribute.name!r} attribute.'
)
[docs] def __init__(self, attribute, **kw):
'''Initialise error.'''
kw.setdefault('details', {}).update(dict(
attribute=attribute
))
super(ImmutableAttributeError, self).__init__(**kw)
[docs]class CollectionError(Error):
'''Raise when an error related to collections occurs.'''
default_message = 'Collection error.'
[docs] def __init__(self, collection, **kw):
'''Initialise error.'''
kw.setdefault('details', {}).update(dict(
collection=collection
))
super(CollectionError, self).__init__(**kw)
[docs]class ImmutableCollectionError(CollectionError):
'''Raise when modification of immutable collection attempted.'''
default_message = (
'Cannot modify value of immutable collection {collection!r}.'
)
[docs]class DuplicateItemInCollectionError(CollectionError):
'''Raise when duplicate item in collection detected.'''
default_message = (
'Item {item!r} already exists in collection {collection!r}.'
)
[docs] def __init__(self, item, collection, **kw):
'''Initialise error.'''
kw.setdefault('details', {}).update(dict(
item=item
))
super(DuplicateItemInCollectionError, self).__init__(collection, **kw)
[docs]class ParseError(Error):
'''Raise when a parsing error occurs.'''
default_message = 'Failed to parse.'
[docs]class EventHubError(Error):
'''Raise when issues related to event hub occur.'''
default_message = 'Event hub error occurred.'
[docs]class EventHubConnectionError(EventHubError):
'''Raise when event hub encounters connection problem.'''
default_message = 'Event hub is not connected.'
[docs]class EventHubPacketError(EventHubError):
'''Raise when event hub encounters an issue with a packet.'''
default_message = 'Invalid packet.'
[docs]class PermissionDeniedError(Error):
'''Raise when permission is denied.'''
default_message = 'Permission denied.'
[docs]class LocationError(Error):
'''Base for errors associated with locations.'''
default_message = 'Unspecified location error'
[docs]class ComponentNotInAnyLocationError(LocationError):
'''Raise when component not available in any location.'''
default_message = 'Component not available in any location.'
[docs]class ComponentNotInLocationError(LocationError):
'''Raise when component(s) not in location.'''
default_message = (
'Component(s) {formatted_components} not found in location {location}.'
)
[docs] def __init__(self, components, location, **kw):
'''Initialise with *components* and *location*.'''
if isinstance(components, ftrack_api.entity.base.Entity):
components = [components]
kw.setdefault('details', {}).update(dict(
components=components,
formatted_components=', '.join(
[str(component) for component in components]
),
location=location
))
super(ComponentNotInLocationError, self).__init__(**kw)
[docs]class ComponentInLocationError(LocationError):
'''Raise when component(s) already exists in location.'''
default_message = (
'Component(s) {formatted_components} already exist in location '
'{location}.'
)
[docs] def __init__(self, components, location, **kw):
'''Initialise with *components* and *location*.'''
if isinstance(components, ftrack_api.entity.base.Entity):
components = [components]
kw.setdefault('details', {}).update(dict(
components=components,
formatted_components=', '.join(
[str(component) for component in components]
),
location=location
))
super(ComponentInLocationError, self).__init__(**kw)
[docs]class AccessorError(Error):
'''Base for errors associated with accessors.'''
default_message = 'Unspecified accessor error'
[docs]class AccessorOperationFailedError(AccessorError):
'''Base for failed operations on accessors.'''
default_message = 'Operation {operation} failed: {error}'
[docs] def __init__(
self, operation='', resource_identifier=None, error=None, **kw
):
kw.setdefault('details', {}).update(dict(
operation=operation,
resource_identifier=resource_identifier,
error=error
))
super(AccessorOperationFailedError, self).__init__(**kw)
[docs]class AccessorUnsupportedOperationError(AccessorOperationFailedError):
'''Raise when operation is unsupported.'''
default_message = 'Operation {operation} unsupported.'
[docs]class AccessorPermissionDeniedError(AccessorOperationFailedError):
'''Raise when permission denied.'''
default_message = (
'Cannot {operation} {resource_identifier}. Permission denied.'
)
[docs]class AccessorResourceIdentifierError(AccessorError):
'''Raise when a error related to a resource_identifier occurs.'''
default_message = 'Resource identifier is invalid: {resource_identifier}.'
[docs] def __init__(self, resource_identifier, **kw):
kw.setdefault('details', {}).update(dict(
resource_identifier=resource_identifier
))
super(AccessorResourceIdentifierError, self).__init__(**kw)
[docs]class AccessorFilesystemPathError(AccessorResourceIdentifierError):
'''Raise when a error related to an accessor filesystem path occurs.'''
default_message = (
'Could not determine filesystem path from resource identifier: '
'{resource_identifier}.'
)
[docs]class AccessorResourceError(AccessorError):
'''Base for errors associated with specific resource.'''
default_message = 'Unspecified resource error: {resource_identifier}'
[docs] def __init__(self, operation='', resource_identifier=None, error=None,
**kw):
kw.setdefault('details', {}).update(dict(
operation=operation,
resource_identifier=resource_identifier
))
super(AccessorResourceError, self).__init__(**kw)
[docs]class AccessorResourceNotFoundError(AccessorResourceError):
'''Raise when a required resource is not found.'''
default_message = 'Resource not found: {resource_identifier}'
[docs]class AccessorParentResourceNotFoundError(AccessorResourceError):
'''Raise when a parent resource (such as directory) is not found.'''
default_message = 'Parent resource is missing: {resource_identifier}'
[docs]class AccessorResourceInvalidError(AccessorResourceError):
'''Raise when a resource is not the right type.'''
default_message = 'Resource invalid: {resource_identifier}'
[docs]class AccessorContainerNotEmptyError(AccessorResourceError):
'''Raise when container is not empty.'''
default_message = 'Container is not empty: {resource_identifier}'
[docs]class StructureError(Error):
'''Base for errors associated with structures.'''
default_message = 'Unspecified structure error'
[docs]class ConnectionClosedError(Error):
'''Raise when attempt to use closed connection detected.'''
default_message = "Connection closed."