Source code for spyne.server.null

#
# spyne - Copyright (C) Spyne contributors.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301
#

"""The ``spyne.server.null`` module contains the NullServer class and its helper
objects.

The name comes from the "null modem connection". Look it up.
"""

import logging
logger = logging.getLogger(__name__)

from spyne import MethodContext

from spyne.client import Factory
from spyne.const.ansi_color import LIGHT_RED
from spyne.const.ansi_color import LIGHT_BLUE
from spyne.const.ansi_color import END_COLOR
from spyne.server import ServerBase


_big_header = ('=' * 40) + LIGHT_RED
_big_footer = END_COLOR + ('=' * 40)
_small_header = ('-' * 20) + LIGHT_BLUE
_small_footer = END_COLOR + ('-' * 20)


class NullServer(ServerBase):
[docs]    """A server that doesn't support any transport at all -- it's implemented
    to test services without having to run a server.

    It implicitly uses the 'sync' auxiliary processing mode.

    Note that:
        1) ``**kwargs`` overwrite ``*args``.
        2) You can do: ::

            logging.getLogger('spyne.server.null').setLevel(logging.CRITICAL)

        to hide context delimiters in logs.
    """

    transport = 'noconn://null.spyne'

    def __init__(self, app, ostr=False, locale='C'):
        super(NullServer, self).__init__(app)

        self.service = _FunctionProxy(self, self.app)
        self.factory = Factory(self.app)
        self.ostr = ostr
        self.locale = locale

    def get_wsdl(self):
[docs]        return self.app.get_interface_document(self.url)

    def set_options(self, **kwargs):
[docs]        self.service.in_header = kwargs.get('soapheaders', self.service.in_header)


class _FunctionProxy(object):
    def __init__(self, server, app):
        self.__app = app
        self.__server = server
        self.in_header = None

    def __getattr__(self, key):
        return _FunctionCall(self.__app, self.__server, key, self.in_header,
                                       self.__server.ostr, self.__server.locale)

    def __getitem__(self, key):
        return self.__getattr__(key)


class _FunctionCall(object):
    def __init__(self, app, server, key, in_header, ostr, locale):
        self.app = app

        self.__key = key
        self.__server = server
        self.__in_header = in_header
        self.__ostr = ostr
        self.__locale = locale

    def __call__(self, *args, **kwargs):
        initial_ctx = MethodContext(self)
        initial_ctx.method_request_string = self.__key
        initial_ctx.in_header = self.__in_header
        initial_ctx.transport.type = NullServer.transport
        initial_ctx.locale = self.__locale

        contexts = self.app.in_protocol.generate_method_contexts(initial_ctx)

        cnt = 0
        retval = None
        logger.warning( "%s start request %s" % (_big_header, _big_footer)  )

        for ctx in contexts:
            # this reconstruction is quite costly. I wonder whether it's a
            # problem though.

            _type_info = ctx.descriptor.in_message._type_info
            ctx.in_object = [None] * len(_type_info)
            for i in range(len(args)):
                ctx.in_object[i] = args[i]

            for i,k in enumerate(_type_info.keys()):
                val = kwargs.get(k, None)
                if val is not None:
                    ctx.in_object[i] = val

            if cnt == 0:
                p_ctx = ctx
            else:
                ctx.descriptor.aux.initialize_context(ctx, p_ctx, error=None)

            # do logging.getLogger('spyne.server.null').setLevel(logging.CRITICAL)
            # to hide the following
            logger.warning( "%s start context %s" % (_small_header, _small_footer) )
            logger.warning( "%r.%r" % (ctx.service_class, ctx.descriptor.function) )
            try:
                self.app.process_request(ctx)
            finally:
                logger.warning( "%s  end context  %s" % (_small_header, _small_footer) )

            if ctx.out_error:
                raise ctx.out_error

            else:
                if len(ctx.descriptor.out_message._type_info) == 0:
                    _retval = None

                elif len(ctx.descriptor.out_message._type_info) == 1:
                    _retval = ctx.out_object[0]

                else:
                    _retval = ctx.out_object

                if cnt == 0 and self.__ostr:
                    self.__server.get_out_string(ctx)
                    _retval = ctx.out_string

            if cnt == 0:
                retval = _retval
            else:
                ctx.close()

            cnt += 1

        p_ctx.close()

        logger.warning( "%s  end request  %s" % (_big_header, _big_footer)  )

        return retval