from typing import Optional
from twirp.exceptions import TwirpServerException
from twirp.errors import Errors as TwirpErrors
[docs]class CatalogException(TwirpServerException):
"""The most generic Catalog platform error"""
def __init__(self, *args, code=TwirpErrors.Unknown, message="", meta={}):
super(CatalogException, self).__init__(code=code,
message=message,
meta=meta)
[docs]class CatalogCanceledException(CatalogException):
"""The operation was cancelled"""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.Canceled,
message=message,
meta=meta)
[docs]class CatalogUnknownException(CatalogException):
"""An unknown error occurred. For example, this can be used when handling errors raised by APIs that do not return any error information."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.Unknown,
message=message,
meta=meta)
[docs]class CatalogInvalidArgumentException(CatalogException):
"""The client specified an invalid argument. This indicates arguments that are invalid regardless of the state of the system (i.e. a malformed file name, required argument, number out of range, etc.)."""
def __init__(self, *args, message, meta={}):
super(CatalogException,
self).__init__(code=TwirpErrors.InvalidArgument,
message=message,
meta=meta)
[docs]class CatalogDeadlineExceededException(CatalogException):
"""Operation expired before completion. For operations that change the state of the system, this error may be returned even if the operation has completed successfully (timeout)."""
def __init__(self, *args, message, meta={}):
super(CatalogException,
self).__init__(code=TwirpErrors.DeadlineExceeded,
message=message,
meta=meta)
[docs]class CatalogNotFoundException(CatalogException):
"""Some requested entity was not found."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.NotFound,
message=message,
meta=meta)
[docs]class CatalogBadRouteException(CatalogException):
"""The requested URL path wasn't routable to a known method. This is returned by generated server code and should not be returned by application code (use "not_found" or "unimplemented" instead)."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.BadRoute,
message=message,
meta=meta)
[docs]class CatalogAlreadyExistsException(CatalogException):
"""An attempt to create an entity failed because one already exists."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.AlreadyExists,
message=message,
meta=meta)
[docs]class CatalogPermissionDeniedException(CatalogException):
"""The caller does not have permission to execute the specified operation. It must not be used if the caller cannot be identified (use "unauthenticated" instead)."""
def __init__(self, *args, message, meta={}):
super(CatalogException,
self).__init__(code=TwirpErrors.PermissionDenied,
message=message,
meta=meta)
[docs]class CatalogUnauthenticatedException(CatalogException):
"The request does not have valid authentication credentials for the operation."
def __init__(self, *args, message, meta={}):
super(CatalogException,
self).__init__(code=TwirpErrors.Unauthenticated,
message=message,
meta=meta)
[docs]class CatalogResourceExhaustedException(CatalogException):
"Some resource has been exhausted or rate-limited, perhaps a per-user quota, or perhaps the entire file system is out of space."
def __init__(self, *args, message, meta={}):
super(CatalogException,
self).__init__(code=TwirpErrors.ResourceExhausted,
message=message,
meta=meta)
[docs]class CatalogFailedPreconditionException(CatalogException):
"""The operation was rejected because the system is not in a state required for the operation's execution. For example, doing an rmdir operation on a directory that is non-empty, or on a non-directory object, or when having conflicting read-modify-write on the same resource."""
def __init__(self, *args, message, meta={}):
super(CatalogException,
self).__init__(code=TwirpErrors.FailedPrecondition,
message=message,
meta=meta)
[docs]class CatalogAbortedException(CatalogException):
"""The operation was aborted, typically due to a concurrency issue like sequencer check failures, transaction aborts, etc."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.Aborted,
message=message,
meta=meta)
[docs]class CatalogOutOfRangeException(CatalogException):
"""The operation was attempted past the valid range. For example, seeking or reading past end of a paginated collection. Unlike "invalid_argument", this error indicates a problem that may be fixed if the system state changes (i.e. adding more items to the collection)."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.OutOfRange,
message=message,
meta=meta)
[docs]class CatalogUnimplementedException(CatalogException):
"""The operation is not implemented or not supported/enabled in this service."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.Unimplemented,
message=message,
meta=meta)
[docs]class CatalogInternalException(CatalogException):
"""When some invariants expected by the underlying system have been broken. In other words, something bad happened in the library or backend service. Twirp specific issues like wire and serialization problems are also reported as "internal" errors."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.Internal,
message=message,
meta=meta)
[docs]class CatalogUnavailableException(CatalogException):
"""The service is currently unavailable. This is most likely a transient condition and may be corrected by retrying with a backoff."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.Unavailable,
message=message,
meta=meta)
[docs]class CatalogDataLossException(CatalogException):
"""The operation resulted in unrecoverable data loss or corruption."""
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.DataLoss,
message=message,
meta=meta)
[docs]class CatalogNoErrorException(CatalogException):
def __init__(self, *args, message, meta={}):
super(CatalogException, self).__init__(code=TwirpErrors.NoError,
message=message,
meta=meta)
def _convert_error(e: Exception) -> CatalogException:
if isinstance(e, TwirpServerException):
err_type = {
TwirpErrors.Canceled: CatalogCanceledException,
TwirpErrors.Unknown: CatalogUnknownException,
TwirpErrors.InvalidArgument: CatalogInvalidArgumentException,
TwirpErrors.Malformed: CatalogMalformedException,
TwirpErrors.DeadlineExceeded: CatalogDeadlineExceededException,
TwirpErrors.NotFound: CatalogNotFoundException,
TwirpErrors.BadRoute: CatalogBadRouteException,
TwirpErrors.AlreadyExists: CatalogAlreadyExistsException,
TwirpErrors.PermissionDenied: CatalogPermissionDeniedException,
TwirpErrors.Unauthenticated: CatalogUnauthenticatedException,
TwirpErrors.ResourceExhausted: CatalogResourceExhaustedException,
TwirpErrors.FailedPrecondition: CatalogFailedPreconditionException,
TwirpErrors.Aborted: CatalogAbortedException,
TwirpErrors.OutOfRange: CatalogOutOfRangeException,
TwirpErrors.Unimplemented: CatalogUnimplementedException,
TwirpErrors.Internal: CatalogInternalException,
TwirpErrors.Unavailable: CatalogUnavailableException,
TwirpErrors.DataLoss: CatalogDataLossException,
TwirpErrors.NoError: CatalogNoErrorException,
}.get(e.code, CatalogException)
return err_type(message=e.message, meta=e.meta)
return e
def _raise_error(e: Exception, message: str):
if isinstance(e, TwirpServerException):
err = _convert_error(e)
k = err.__class__
raise k(message="{}. Reason: {}".format(message, err.message)) from err
else:
raise e