Source code for libcloud.utils.loggingconnection

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import absolute_import

    import simplejson as json  # type: ignore
except Exception:
    import json  # type: ignore

from pipes import quote as pquote
from xml.dom.minidom import parseString

import os

from libcloud.common.base import (LibcloudConnection,
from libcloud.utils.py3 import _real_unicode as u
from libcloud.utils.py3 import ensure_string

from libcloud.utils.misc import lowercase_keys

[docs]class LoggingConnection(LibcloudConnection): """ Debug class to log all HTTP(s) requests as they could be made with the curl command. :cvar log: file-like object that logs entries are written to. """ protocol = 'https' log = None http_proxy_used = False def _log_response(self, r): rv = "# -------- begin %d:%d response ----------\n" % (id(self), id(r)) ht = "" v = r.version if r.version == 10: v = "HTTP/1.0" if r.version == 11: v = "HTTP/1.1" ht += "%s %s %s\r\n" % (v, r.status, r.reason) body = for h in r.getheaders(): ht += "%s: %s\r\n" % (h[0].title(), h[1]) ht += "\r\n" headers = lowercase_keys(dict(r.getheaders())) content_type = headers.get('content-type', None) pretty_print = os.environ.get('LIBCLOUD_DEBUG_PRETTY_PRINT_RESPONSE', False) if pretty_print and content_type == 'application/json': try: body = json.loads(ensure_string(body)) body = json.dumps(body, sort_keys=True, indent=4) except Exception: # Invalid JSON or server is lying about content-type pass elif pretty_print and content_type in ['text/xml', 'application/xml']: try: elem = parseString(body.decode('utf-8')) body = elem.toprettyxml() except Exception: # Invalid XML pass ht += ensure_string(body) rv += ht rv += ("\n# -------- end %d:%d response ----------\n" % (id(self), id(r))) return rv def _log_curl(self, method, url, body, headers): cmd = ["curl"] if self.http_proxy_used: if self.proxy_username and self.proxy_password: proxy_url = '%s://%s:%s@%s:%s' % (self.proxy_scheme, self.proxy_username, self.proxy_password, self.proxy_host, self.proxy_port) else: proxy_url = '%s://%s:%s' % (self.proxy_scheme, self.proxy_host, self.proxy_port) proxy_url = pquote(proxy_url) cmd.extend(['--proxy', proxy_url]) cmd.extend(['-i']) if method.lower() == 'head': # HEAD method need special handling cmd.extend(["--head"]) else: cmd.extend(["-X", pquote(method)]) for h in headers: cmd.extend(["-H", pquote("%s: %s" % (h, headers[h]))]) cert_file = getattr(self, 'cert_file', None) if cert_file: cmd.extend(["--cert", pquote(cert_file)]) # TODO: in python 2.6, body can be a file-like object. if body is not None and len(body) > 0: if isinstance(body, (bytearray, bytes)): body = body.decode('utf-8') cmd.extend(["--data-binary", pquote(body)]) cmd.extend(["--compress"]) cmd.extend([pquote("%s%s" % (, url))]) return " ".join(cmd)
[docs] def getresponse(self): original_response = LibcloudConnection.getresponse(self) if self.log is not None: rv = self._log_response(HttpLibResponseProxy(original_response)) self.log.write(u(rv + "\n")) self.log.flush() return original_response
[docs] def request(self, method, url, body=None, headers=None, **kwargs): headers.update({'X-LC-Request-ID': str(id(self))}) if self.log is not None: pre = "# -------- begin %d request ----------\n" % id(self) self.log.write(u(pre + self._log_curl(method, url, body, headers) + "\n")) self.log.flush() return LibcloudConnection.request(self, method, url, body, headers)