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 get_http_connection
  36 from kamaki.clients.connection import HTTPConnection, HTTPResponse
  37 from kamaki.clients.connection.errors import HTTPConnectionError
  38 from kamaki.clients.connection.errors import HTTPResponseError
  39 from socket import gaierror, error
  40
  41 from json import loads
  42
  43 from time import sleep
  44 from httplib import ResponseNotReady
  45
  46
  47 class KamakiHTTPResponse(HTTPResponse):
  48
  49     def _get_response(self):
  50         if self.prefetched:
  51             return
  52
  53         ready = False
  54         while not ready:
  55             try:
  56                 r = self.request.getresponse()
  57             except ResponseNotReady:
  58                 sleep(0.001)
  59                 continue
  60             break
  61         self.prefetched = True
  62         headers = {}
  63         for k, v in r.getheaders():
  64             headers.update({k: v})
  65         self.headers = headers
  66         self.content = r.read()
  67         self.status_code = r.status
  68         self.status = r.reason
  69         self.request.close()
  70
  71     @property
  72     def text(self):
  73         """
  74         :returns: (str) content
  75         """
  76         self._get_response()
  77         return unicode(self._content)
  78
  79     @text.setter
  80     def test(self, v):
  81         pass
  82
  83     @property
  84     def json(self):
  85         """
  86         :returns: (dict) the json-formated content
  87
  88         :raises HTTPResponseError: if content is not json formated
  89         """
  90         self._get_response()
  91         try:
  92             return loads(self._content)
  93         except ValueError as err:
  94             HTTPResponseError('Response not formated in JSON - %s' % err)
  95
  96     @json.setter
  97     def json(self, v):
  98         pass
  99
 100     def release(self):
 101         """ Release the connection. Should always be called if the response
 102         content hasn't been used.
 103         """
 104         if not self.prefetched:
 105             self.request.close()
 106
 107
 108 class KamakiHTTPConnection(HTTPConnection):
 109
 110     def _retrieve_connection_info(self, extra_params={}):
 111         """
 112         :param extra_params: (dict) key:val for url parameters
 113
 114         :returns: (scheme, netloc, url?with&params)
 115         """
 116         if self.url:
 117             url = self.url if self.url[-1] == '/' else (self.url + '/')
 118         else:
 119             url = 'http://127.0.0.1'
 120         if self.path:
 121             url += self.path[1:] if self.path[0] == '/' else self.path
 122         params = dict(self.params)
 123         for k, v in extra_params.items():
 124             params[k] = v
 125         for i, (key, val) in enumerate(params.items()):
 126             param_str = ('?' if i == 0 else '&') + unicode(key)
 127             if val is not None:
 128                 param_str += '=' + unicode(val)
 129             url += param_str
 130
 131         parsed = urlparse(url)
 132         self.url = url
 133         self.path = parsed.path if parsed.path else '/'
 134         self.path += '?%s' % parsed.query if parsed.query else ''
 135         return (parsed.scheme, parsed.netloc)
 136
 137     def perform_request(self,
 138         method=None,
 139         data=None,
 140         async_headers={},
 141         async_params={}):
 142         """
 143         :param method: (str) http method ('get', 'post', etc.)
 144
 145         :param data: (binary object)
 146
 147         :param async_headers: (dict) key:val headers that are used only for one
 148             request instance as opposed to self.headers, which remain to be
 149             used by following or parallel requests
 150
 151         :param async_params: (dict) key:val url parameters that are used only
 152             for one request instance as opposed to self.params, which remain to
 153             be used by following or parallel requests
 154
 155         :returns: (KamakiHTTPResponse) a response object
 156
 157         :raises HTTPConnectionError: Connection failures
 158         """
 159         (scheme, netloc) = self._retrieve_connection_info(
 160             extra_params=async_params)
 161         headers = dict(self.headers)
 162         for k, v in async_headers.items():
 163             headers[k] = v
 164
 165         #de-unicode headers to prepare them for http
 166         http_headers = {}
 167         for k, v in headers.items():
 168             http_headers[str(k)] = str(v)
 169
 170         #get connection from pool
 171         conn = get_http_connection(netloc=netloc, scheme=scheme)
 172         try:
 173             #Be carefull, all non-body variables should not be unicode
 174             conn.request(method=str(method.upper()),
 175                 url=str(self.path),
 176                 headers=http_headers,
 177                 body=data)
 178         except IOError as ioe:
 179             raise HTTPConnectionError(
 180                 'Cannot connect to %s: %s' % (self.url, ioe.strerror),
 181                 errno=ioe.errno)
 182         except Exception as err:
 183             from traceback import format_stack
 184             from kamaki.clients import recvlog
 185             recvlog.debug('\n'.join(['%s' % type(err)] + format_stack()))
 186             conn.close()
 187             raise
 188         return KamakiHTTPResponse(conn)