Source code for dace.http

from StringIO import StringIO
from io import BytesIO
import json
from dace.error import DaceRetryRequestedError
from twisted.python.failure import Failure
from twisted.web._newclient import RequestTransmissionFailed
from twisted.internet.task import TaskStopped
from twisted.python.components import registerAdapter, getRegistry
from twisted.web.client import FileBodyProducer
from zope.interface import implementer, declarations
from twisted.web.iweb import IBodyProducer
from treq.client import HTTPClient


[docs]class Http(object):
    def __init__(self, agent, reactor):
        self.agent = agent
        self.reactor = reactor

[docs]    def head(self, url, **kwargs):
        """
        Make a ``HEAD`` request.

        See :py:func:`treq.request`
        """
        return self.request('HEAD', url, **kwargs)

[docs]    def get(self, url, headers=None, **kwargs):
        """
        Make a ``GET`` request.

        See :py:func:`treq.request`
        """
        return self.request('GET', url, headers=headers, **kwargs)

[docs]    def post(self, url, data=None, **kwargs):
        """
        Make a ``POST`` request.

        See :py:func:`treq.request`
        """
        return self.request('POST', url, data=data, **kwargs)

[docs]    def put(self, url, data=None, **kwargs):
        """
        Make a ``PUT`` request.

        See :py:func:`treq.request`
        """
        return self.request('PUT', url, data=data, **kwargs)

[docs]    def patch(self, url, data=None, **kwargs):
        """
        Make a ``PATCH`` request.

        See :py:func:`treq.request`
        """
        return self.request('PATCH', url, data=data, **kwargs)

[docs]    def delete(self, url, **kwargs):
        """
        Make a ``DELETE`` request.

        See :py:func:`treq.request`
        """
        return self.request('DELETE', url, **kwargs)

[docs]    def request(self, method, url, **kwargs):
        """
        Make an HTTP request.

        :param str method: HTTP method. Example: ``'GET'``, ``'HEAD'``. ``'PUT'``,
             ``'POST'``.
        :param str url: http or https URL, which may include query arguments.

        :param headers: Optional HTTP Headers to send with this request.
        :type headers: Headers or None

        :param params: Optional parameters to be append as the query string to
            the URL, any query string parameters in the URL already will be
            preserved.

        :type params: dict w/ str or list/tuple of str values, list of 2-tuples, or
            None.

        :param data: Optional request body.
        :type data: str, file-like, IBodyProducer, or None

        :param reactor: Optional twisted reactor.

        :param bool persistent: Use persistent HTTP connections.  Default: ``True``
        :param bool allow_redirects: Follow HTTP redirects.  Default: ``True``

        :param auth: HTTP Basic Authentication information.
        :type auth: tuple of ('username', 'password').

        :param int timeout: Request timeout seconds. If a response is not
            received within this timeframe, a connection is aborted with
            ``CancelledError``.

        :rtype: Deferred that fires with an IResponse provider.

        """
        kwargs['reactor'] = self.reactor

        def eb(failure):
            if failure.check(RequestTransmissionFailed):
                return Failure(DaceRetryRequestedError())
            else:
                return failure

        d = HTTPClient(self.agent).request(method, url, **kwargs)
        d.addErrback(eb)
        return d


class JsonBody(object):
    """
    Wraps an object to be serialized as json
    """
    def __init__(self, data):
        super(JsonBody, self).__init__()
        self.data = data


@implementer(IBodyProducer)
class StfuFileBodyProducer(FileBodyProducer):
    """File producer that consumes TaskStopped error.

    See http://twistedmatrix.com/trac/ticket/6528 for details.
    """
    def stopProducing(self):
        try:
            super(StfuFileBodyProducer, self).stopProducing()
        except TaskStopped:
            pass


def _from_json_body(json_body):
    return StfuFileBodyProducer(StringIO(json.dumps(json_body.data)))

def _from_bytes(orig_bytes):
    return StfuFileBodyProducer(StringIO(orig_bytes))

def _from_file(orig_file):
    return StfuFileBodyProducer(orig_file)


def _changeAdapter(adapterFactory, origInterface, interfaceClass):
    """Unregister previously registered adapter and register a new one"""
    getRegistry().register(
        [declarations.implementedBy(origInterface)],
        interfaceClass,
        '',
        None
    )
    registerAdapter(adapterFactory, origInterface, interfaceClass)


_changeAdapter(_from_bytes, str, IBodyProducer)
_changeAdapter(_from_file, file, IBodyProducer)
_changeAdapter(_from_file, StringIO, IBodyProducer)
_changeAdapter(_from_file, BytesIO, IBodyProducer)
registerAdapter(_from_json_body, JsonBody, IBodyProducer)

__all__ = ['Http']