code.grnet.gr Git - kamaki/blob - kamaki/clients/connection/kamakicon.py

   1 # Copyright 2012 GRNET S.A. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or
   4 # without modification, are permitted provided that the following
   5 # conditions are met:
   6 #
   7 #   1. Redistributions of source code must retain the above
   8 #      copyright notice, self.list of conditions and the following
   9 #      disclaimer.
  10 #
  11 #   2. Redistributions in binary form must reproduce the above
  12 #      copyright notice, self.list of conditions and the following
  13 #      disclaimer in the documentation and/or other materials
  14 #      provided with the distribution.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  17 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  19 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  20 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  23 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  24 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  25 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  26 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27 # POSSIBILITY OF SUCH DAMAGE.
  28 #
  29 # The views and conclusions contained in the software and
  30 # documentation are those of the authors and should not be
  31 # interpreted as representing official policies, either expressed
  32 # or implied, of GRNET S.A.
  33
  34 from urlparse import urlparse
  35 from objpool.http import PooledHTTPConnection
  36 from traceback import format_stack
  37
  38 from kamaki.clients.connection import HTTPConnection, HTTPResponse
  39 from kamaki.clients.connection.errors import HTTPConnectionError
  40 from kamaki.clients.connection.errors import HTTPResponseError
  41
  42 from json import loads
  43
  44 from time import sleep
  45 from httplib import ResponseNotReady
  46
  47
  48 class KamakiHTTPResponse(HTTPResponse):
  49     """The request is created only if there is demand for the response"""
  50
  51     def __init__(
  52         self, netloc, scheme,
  53         method='GET', url='127.0.0.1', headers={}, body=None, poolsize=None):
  54         self.netloc, self.scheme = netloc, scheme
  55         self.method, self.url, self.http_headers = method, url, headers
  56         self.body = body
  57         self.poolsize = poolsize
  58
  59     def _get_response(self):
  60         try:
  61             if self.prefetched:
  62                 return
  63         except AttributeError:
  64             pass
  65
  66         try:
  67             with PooledHTTPConnection(
  68                     self.netloc, self.scheme,
  69                     size=self.poolsize) as conn:
  70                 super(KamakiHTTPResponse, self).__init__(conn)
  71                 conn.request(
  72                     method=str(self.method),
  73                     url=str(self.url),
  74                     headers=self.http_headers,
  75                     body=self.body)
  76                 while True:
  77                     try:
  78                         r = self.request.getresponse()
  79                     except ResponseNotReady:
  80                         sleep(0.001)
  81                         continue
  82                     break
  83                 self.prefetched = True
  84                 headers = {}
  85                 for k, v in r.getheaders():
  86                     headers.update({k: v})
  87                 self.headers = headers
  88                 self.content = r.read()
  89                 self.status_code = r.status
  90                 self.status = r.reason
  91         except IOError as ioe:
  92             raise HTTPConnectionError(
  93                 'Cannot connect to %s: %s' % (self.url, ioe.strerror),
  94                 errno=ioe.errno)
  95         except Exception as err:
  96             from kamaki.clients import recvlog
  97             recvlog.debug('\n'.join(['%s' % type(err)] + format_stack()))
  98             raise HTTPConnectionError(
  99                 'Failed to handle connection to %s %s' % (self.url, err),
 100                 errno=-1)
 101
 102     @property
 103     def text(self):
 104         """
 105         :returns: (str) content
 106         """
 107         self._get_response()
 108         return '%s' % self._content
 109
 110     @text.setter
 111     def text(self, v):
 112         pass
 113
 114     @property
 115     def json(self):
 116         """
 117         :returns: (dict) the json-formated content
 118
 119         :raises HTTPResponseError: if content is not json formated
 120         """
 121         self._get_response()
 122         try:
 123             return loads(self._content)
 124         except ValueError as err:
 125             HTTPResponseError('Response not formated in JSON - %s' % err)
 126
 127     @json.setter
 128     def json(self, v):
 129         pass
 130
 131     def release(self):
 132         (self.netloc, self.scheme, self.poolsize) = (None, None, None)
 133
 134
 135 class KamakiHTTPConnection(HTTPConnection):
 136
 137     def _retrieve_connection_info(self, extra_params={}):
 138         """
 139         :param extra_params: (dict) key:val for url parameters
 140
 141         :returns: (scheme, netloc, url?with&params)
 142         """
 143         if self.url:
 144             url = self.url if self.url[-1] == '/' else (self.url + '/')
 145         else:
 146             url = 'http://127.0.0.1'
 147         if self.path:
 148             url += self.path[1:] if self.path[0] == '/' else self.path
 149         params = dict(self.params)
 150         for k, v in extra_params.items():
 151             params[k] = v
 152         for i, (key, val) in enumerate(params.items()):
 153             param_str = '%s%s' % ('?' if i == 0 else '&', key)
 154             if val is not None:
 155                 param_str += '=%s' % val
 156             url += param_str
 157
 158         parsed = urlparse(url)
 159         self.url = url
 160         self.path = parsed.path if parsed.path else '/'
 161         self.path += '?%s' % parsed.query if parsed.query else ''
 162         return (parsed.scheme, parsed.netloc)
 163
 164     def perform_request(
 165             self,
 166             method=None, data=None, async_headers={}, async_params={}):
 167         """
 168         :param method: (str) http method ('get', 'post', etc.)
 169
 170         :param data: (binary object)
 171
 172         :param async_headers: (dict) key:val headers that are used only for one
 173             request instance as opposed to self.headers, which remain to be
 174             used by following or parallel requests
 175
 176         :param async_params: (dict) key:val url parameters that are used only
 177             for one request instance as opposed to self.params, which remain to
 178             be used by following or parallel requests
 179
 180         :returns: (KamakiHTTPResponse) a response object
 181
 182         :raises HTTPConnectionError: Connection failures
 183         """
 184         (scheme, netloc) = self._retrieve_connection_info(
 185             extra_params=async_params)
 186         headers = dict(self.headers)
 187         for k, v in async_headers.items():
 188             headers[k] = v
 189
 190         #de-unicode headers to prepare them for http
 191         http_headers = {}
 192         for k, v in headers.items():
 193             http_headers[str(k)] = str(v)
 194
 195         return KamakiHTTPResponse(
 196             netloc, scheme, method.upper(), self.path,
 197             headers=http_headers, body=data, poolsize=self.poolsize)