manyi/venv/Lib/site-packages/hyper/cli.py

# -*- coding: utf-8 -*-
"""
hyper/cli
~~~~~~~~~

Command line interface for Hyper inspired by Httpie.
"""
import json
import locale
import logging
import sys
from argparse import ArgumentParser, RawTextHelpFormatter
from argparse import OPTIONAL, ZERO_OR_MORE
from pprint import pformat
from textwrap import dedent

from hyper import HTTPConnection, HTTP20Connection
from hyper import __version__
from hyper.compat import is_py2, urlencode, urlsplit, write_to_stdout
from hyper.common.util import to_host_port_tuple


log = logging.getLogger('hyper')

PREFERRED_ENCODING = locale.getpreferredencoding()

# Various separators used in args
SEP_HEADERS = ':'
SEP_QUERY = '=='
SEP_DATA = '='

SEP_GROUP_ITEMS = [
    SEP_HEADERS,
    SEP_QUERY,
    SEP_DATA,
]


class KeyValue(object):
    """Base key-value pair parsed from CLI."""

    def __init__(self, key, value, sep, orig):
        self.key = key
        self.value = value
        self.sep = sep
        self.orig = orig


class KeyValueArgType(object):
    """A key-value pair argument type used with `argparse`.

    Parses a key-value arg and constructs a `KeyValue` instance.
    Used for headers, form data, and other key-value pair types.
    This class is inspired by httpie and implements simple tokenizer only.
    """
    def __init__(self, *separators):
        self.separators = separators

    def __call__(self, string):
        for sep in self.separators:
            splitted = string.split(sep, 1)
            if len(splitted) == 2:
                key, value = splitted
                return KeyValue(key, value, sep, string)


def make_positional_argument(parser):
    parser.add_argument(
        'method', metavar='METHOD', nargs=OPTIONAL, default='GET',
        help=dedent("""
        The HTTP method to be used for the request
        (GET, POST, PUT, DELETE, ...).
        """))
    parser.add_argument(
        '_url', metavar='URL',
        help=dedent("""
        The scheme defaults to 'https://' if the URL does not include one.
        """))
    parser.add_argument(
        'items',
        metavar='REQUEST_ITEM',
        nargs=ZERO_OR_MORE,
        type=KeyValueArgType(*SEP_GROUP_ITEMS),
        help=dedent("""
        Optional key-value pairs to be included in the request.
        The separator used determines the type:

        ':' HTTP headers:

            Referer:http://httpie.org  Cookie:foo=bar  User-Agent:bacon/1.0

        '==' URL parameters to be appended to the request URI:

            search==hyper

        '=' Data fields to be serialized into a JSON object:

            name=Hyper  language=Python  description='CLI HTTP client'
        """))


def make_troubleshooting_argument(parser):
    parser.add_argument(
        '--version', action='version', version=__version__,
        help='Show version and exit.')
    parser.add_argument(
        '--debug', action='store_true', default=False,
        help='Show debugging information (loglevel=DEBUG)')
    parser.add_argument(
        '--h2', action='store_true', default=False,
        help="Do HTTP/2 directly, skipping plaintext upgrade and ignoring "
             "NPN/ALPN."
    )


def split_host_and_port(hostname):
    if ':' in hostname:
        return to_host_port_tuple(hostname, default_port=443)
    return hostname, None


class UrlInfo(object):
    def __init__(self):
        self.fragment = None
        self.host = 'localhost'
        self.netloc = None
        self.path = '/'
        self.port = 443
        self.query = None
        self.scheme = 'https'
        self.secure = False


def set_url_info(args):
    info = UrlInfo()
    _result = urlsplit(args._url)
    for attr in vars(info).keys():
        value = getattr(_result, attr, None)
        if value:
            setattr(info, attr, value)

    if info.scheme == 'http' and not _result.port:
        info.port = 80

    # Set the secure arg is the scheme is HTTPS, otherwise do unsecured.
    info.secure = info.scheme == 'https'

    if info.netloc:
        hostname, _ = split_host_and_port(info.netloc)
        info.host = hostname  # ensure stripping port number
    else:
        if _result.path:
            _path = _result.path.split('/', 1)
            hostname, port = split_host_and_port(_path[0])
            info.host = hostname
            if info.path == _path[0]:
                info.path = '/'
            elif len(_path) == 2 and _path[1]:
                info.path = '/' + _path[1]
            if port is not None:
                info.port = port

    log.debug('Url Info: %s', vars(info))
    args.url = info


def set_request_data(args):
    body, headers, params = {}, {}, {}
    for i in args.items:
        if i.sep == SEP_HEADERS:
            if i.key:
                headers[i.key] = i.value
            else:
                # when overriding a HTTP/2 special header there will be a
                # leading colon, which tricks the command line parser into
                # thinking the header is empty
                k, v = i.value.split(':', 1)
                headers[':' + k] = v
        elif i.sep == SEP_QUERY:
            params[i.key] = i.value
        elif i.sep == SEP_DATA:
            value = i.value
            if is_py2:  # pragma: no cover
                value = value.decode(PREFERRED_ENCODING)
            body[i.key] = value

    if params:
        args.url.path += '?' + urlencode(params)

    if body:
        content_type = 'application/json'
        headers.setdefault('content-type', content_type)
        args.body = json.dumps(body)

    if args.method is None:
        args.method = 'POST' if args.body else 'GET'

    args.method = args.method.upper()
    args.headers = headers


def parse_argument(argv=None):
    parser = ArgumentParser(formatter_class=RawTextHelpFormatter)
    parser.set_defaults(body=None, headers={})
    make_positional_argument(parser)
    make_troubleshooting_argument(parser)
    args = parser.parse_args(sys.argv[1:] if argv is None else argv)

    if args.debug:
        handler = logging.StreamHandler()
        handler.setLevel(logging.DEBUG)
        log.addHandler(handler)
        log.setLevel(logging.DEBUG)

    set_url_info(args)
    set_request_data(args)
    return args


def get_content_type_and_charset(response):
    charset = 'utf-8'
    content_type = response.headers.get('content-type')
    if content_type is None:
        return 'unknown', charset

    content_type = content_type[0].decode('utf-8').lower()
    type_and_charset = content_type.split(';', 1)
    ctype = type_and_charset[0].strip()
    if len(type_and_charset) == 2:
        charset = type_and_charset[1].strip().split('=')[1]

    return ctype, charset


def request(args):
    if not args.h2:
        conn = HTTPConnection(
            args.url.host, args.url.port, secure=args.url.secure
        )
    else:  # pragma: no cover
        conn = HTTP20Connection(
            args.url.host,
            args.url.port,
            secure=args.url.secure,
            force_proto='h2'
        )

    conn.request(args.method, args.url.path, args.body, args.headers)
    response = conn.get_response()
    log.debug('Response Headers:\n%s', pformat(response.headers))
    ctype, charset = get_content_type_and_charset(response)
    data = response.read()
    return data


def main(argv=None):
    args = parse_argument(argv)
    log.debug('Commandline Argument: %s', args)
    data = request(args)
    write_to_stdout(data)


if __name__ == '__main__':  # pragma: no cover
    main()