Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class MissingFieldError(InvalidInputError):
"""Raised when a mandatory value is missing."""
CODE = 'Client.InvalidInput'
def __init__(self, field_name, message="Field '%s' is missing."):
try:
message = message % (field_name,)
except TypeError:
pass
super(MissingFieldError, self).__init__(self.CODE, message)
class ValidationError(Fault):
"""Raised when the input stream does not adhere to type constraints."""
CODE = 'Client.ValidationError'
def __init__(self, obj, custom_msg='The value %r could not be validated.'):
try:
msg = custom_msg % (obj,)
except TypeError:
msg = custom_msg
super(ValidationError, self).__init__(self.CODE, msg)
class InternalError(Fault):
"""Raised to communicate server-side errors."""
super(PublicKeyError, self).__init__(
faultstring='Value %r not found' % value)
class AuthenticationError(Fault):
__namespace__ = 'spyne.examples.authentication'
def __init__(self, user_name):
# TODO: self.transport.http.resp_code = HTTP_401
super(AuthenticationError, self).__init__(
faultcode='Client.AuthenticationError',
faultstring='Invalid authentication request for %r' % user_name)
class AuthorizationError(Fault):
__namespace__ = 'spyne.examples.authentication'
def __init__(self):
# TODO: self.transport.http.resp_code = HTTP_401
super(AuthorizationError, self).__init__(
faultcode='Client.AuthorizationError',
faultstring='You are not authozied to access this resource.')
class SpyneDict(dict):
def __getitem__(self, key):
try:
return dict.__getitem__(self, key)
except KeyError:
raise PublicKeyError(key)
import logging
logger = logging.getLogger(__name__)
from spyne.util import six
import msgpack
from spyne.model.fault import Fault
from spyne.protocol.dictdoc import HierDictDocument
from spyne.model.primitive import Double
from spyne.model.primitive import Boolean
from spyne.model.primitive import Integer
class MessagePackDecodeError(Fault):
def __init__(self, data=None):
super(MessagePackDecodeError, self).__init__("Client.MessagePackDecodeError", data)
class MessagePackDocument(HierDictDocument):
"""An integration class for the msgpack protocol."""
mime_type = 'application/x-msgpack'
type = set(HierDictDocument.type)
type.add('msgpack')
# flags to be used in tests
_decimal_as_string = True
_huge_numbers_as_string = True
Fault.__init__(
self,
faultcode="ElbeNotLoggedIn",
faultstring="Not authenticated ! "
"Cant let you perform this command.")
class SoapElbeNotAuthorized(Fault):
def __init__(self):
Fault.__init__(
self,
faultcode="ElbeNotAuthorized",
faultstring="Not Authorized ! Cant let you perform this command.")
class SoapElbeValidationError(Fault):
def __init__(self, exc):
Fault.__init__(
self,
faultcode="ElbeValidationError",
faultstring=exc.__repr__())
class SoapElbeInvalidState(Fault):
def __init__(self):
Fault.__init__(self, faultcode="ElbeInvalidState",
faultstring="Project is Busy ! Operation Invalid")
def soap_faults(func):
""" decorator, which wraps Exceptions to the proper
Soap Faults, and raises these.
def _eb_deferred(retval):
p_ctx.out_error = retval.value
if not issubclass(retval.type, Fault):
retval.printTraceback()
tpt.get_out_string(p_ctx)
self.sendFrame(opcode, ''.join(p_ctx.out_string), fin)
p_ctx.close()
from spyne.model.primitive import Mandatory
from spyne.model.primitive import String
from spyne.protocol.soap import Soap11
from spyne.server.wsgi import WsgiApplication
from spyne.service import Service
class PublicKeyError(Fault):
__namespace__ = 'spyne.examples.authentication'
def __init__(self, value):
super(PublicKeyError, self).__init__(
faultstring='Value %r not found' % value)
class AuthenticationError(Fault):
__namespace__ = 'spyne.examples.authentication'
def __init__(self, user_name):
# TODO: self.transport.http.resp_code = HTTP_401
super(AuthenticationError, self).__init__(
faultcode='Client.AuthenticationError',
faultstring='Invalid authentication request for %r' % user_name)
class AuthorizationError(Fault):
__namespace__ = 'spyne.examples.authentication'
def __init__(self):
# TODO: self.transport.http.resp_code = HTTP_401
binary.ByteArray(encoding='base64'),
binary.ByteArray(encoding='hex'),
],
[
primitive.Point(2), primitive.Point(3),
primitive.Line(2), primitive.Line(3),
primitive.Polygon(2), primitive.Polygon(3),
primitive.MultiPoint(2), primitive.MultiPoint(3),
primitive.MultiLine(2), primitive.MultiLine(3),
primitive.MultiPolygon(2), primitive.MultiPolygon(3),
]
)
if isclass(cls)
and issubclass(cls, ModelBase)
and not issubclass(cls, (Fault, AnyHtml))
and not cls in (ModelBase,)
])
if __name__ == '__main__':
from pprint import pprint
pprint(TYPE_MAP)
class RespawnError(ResourceNotFoundError):
pass
class ResourceAlreadyExistsError(Fault):
"""Raised when requested resource already exists on server side."""
CODE = 'Client.ResourceAlreadyExists'
def __init__(self, fault_object, fault_string="Resource %r already exists"):
super(ResourceAlreadyExistsError, self) \
.__init__(self.CODE, fault_string % fault_object)
class Redirect(Fault):
"""Raised when client needs to make another request for the same
resource."""
CODE = 'Client.Redirect'
def __init__(self, ctx, location, orig_exc=None):
super(Redirect, self).__init__(self.CODE, faultstring=location)
self.ctx = ctx
self.location = location
self.orig_exc = orig_exc
def do_redirect(self):
raise NotImplementedError()
def serialize(self, ctx, message):
assert message in (self.REQUEST, self.RESPONSE)
self.event_manager.fire_event('before_serialize', ctx)
# construct the soap response, and serialize it
nsmap = self.app.interface.nsmap
ctx.out_document = {
"ver": self.version,
}
if ctx.out_error is not None:
ctx.out_document[self.FAULT] = Fault.to_dict(Fault, ctx.out_error)
else:
if message is self.REQUEST:
header_message_class = ctx.descriptor.in_header
body_message_class = ctx.descriptor.in_message
elif message is self.RESPONSE:
header_message_class = ctx.descriptor.out_header
body_message_class = ctx.descriptor.out_message
# assign raw result to its wrapper, result_message
out_type_info = body_message_class._type_info
out_object = body_message_class()
keys = iter(out_type_info)
values = iter(ctx.out_object)
from spyne.model.fault import Fault
class InvalidCredentialsError(Fault):
"""Raised when requested resource is forbidden."""
CODE = 'Client.InvalidCredentialsError'
STR = "You do not have permission to access this resource."
def __init__(self, fault_string=STR, params=None):
super(InvalidCredentialsError, self) \
.__init__(self.CODE, fault_string, detail=params)
class RequestTooLongError(Fault):
"""Raised when request is too long."""
CODE = 'Client.RequestTooLong'
def __init__(self, faultstring="Request too long"):
super(RequestTooLongError, self).__init__(self.CODE, faultstring)
class RequestNotAllowed(Fault):
"""Raised when request is incomplete."""
CODE = 'Client.RequestNotAllowed'
def __init__(self, faultstring=""):
super(RequestNotAllowed, self).__init__(self.CODE, faultstring)