1
0
mirror of https://github.com/httpie/cli.git synced 2024-11-24 08:22:22 +02:00

Initial --download implementation (#104).

Closes #127
This commit is contained in:
Jakub Roztocil 2013-02-26 15:12:33 +01:00
parent f0c42cd089
commit 8e6c765be2
12 changed files with 631 additions and 214 deletions

View File

@ -16,6 +16,10 @@ for **testing, debugging**, and generally **interacting** with HTTP servers.
:height: 835
:align: center
------
.. image:: https://raw.github.com/claudiatd/httpie-artwork/master/images/httpie_logo_simple.png
:alt: HTTPie logo
:align: center
@ -158,6 +162,13 @@ Download a file and save it via `redirected output`_:
$ http example.org/file > file
Download a file ``wget`` style:
.. code-block:: bash
$ http --download example.org/file
Use named `sessions`_ to make certain aspects or the communication persistent
between requests to the same host:
@ -811,6 +822,15 @@ by adding the following to your ``~/.bash_profile``:
}
=============
Download Mode
=============
HTTPie features a download mode, in which a download progress bar is shown,
and the response body is saved to a file. You can enable this mode
with the ``--download`` flag.
==================
Streamed Responses
==================
@ -1081,6 +1101,8 @@ Changelog
*You can click a version name to see a diff with the previous one.*
* `0.5.0-alpha`_
* Added ``--download`` mode.
* `0.4.1`_ (2013-02-26)
* Fixed ``setup.py``.
* `0.4.0`_ (2013-02-22)

View File

@ -3,7 +3,7 @@ HTTPie - a CLI, cURL-like tool for humans.
"""
__author__ = 'Jakub Roztocil'
__version__ = '0.4.1'
__version__ = '0.5.0-alpha'
__licence__ = 'BSD'

View File

@ -1,15 +1,12 @@
"""CLI arguments definition.
NOTE: the CLI interface may change before reaching v1.0.
TODO: make the options config friendly, i.e., no mutually exclusive groups to
allow options overwriting.
"""
from argparse import FileType, OPTIONAL, ZERO_OR_MORE, SUPPRESS
from . import __doc__
from . import __version__
from .compat import is_windows
from .sessions import DEFAULT_SESSIONS_DIR, Session
from .output import AVAILABLE_STYLES, DEFAULT_STYLE
from .input import (Parser, AuthCredentialsArgType, KeyValueArgType,
@ -45,7 +42,8 @@ positional = parser.add_argument_group(
''')
)
positional.add_argument(
'method', metavar='METHOD',
'method',
metavar='METHOD',
nargs=OPTIONAL,
default=None,
help=_('''
@ -57,14 +55,16 @@ positional.add_argument(
''')
)
positional.add_argument(
'url', metavar='URL',
'url',
metavar='URL',
help=_('''
The protocol defaults to http:// if the
URL does not include one.
''')
)
positional.add_argument(
'items', metavar='REQUEST ITEM',
'items',
metavar='REQUEST ITEM',
nargs=ZERO_OR_MORE,
type=KeyValueArgType(*SEP_GROUP_ITEMS),
help=_('''
@ -90,7 +90,8 @@ content_type = parser.add_argument_group(
)
content_type.add_argument(
'--json', '-j', action='store_true',
'--json', '-j',
action='store_true',
help=_('''
(default) Data items from the command
line are serialized as a JSON object.
@ -99,7 +100,8 @@ content_type.add_argument(
''')
)
content_type.add_argument(
'--form', '-f', action='store_true',
'--form', '-f',
action='store_true',
help=_('''
Data items from the command line are serialized as form fields.
The Content-Type is set to application/x-www-form-urlencoded
@ -117,20 +119,9 @@ content_type.add_argument(
output_processing = parser.add_argument_group(title='Output processing')
output_processing.add_argument(
'--output', '-o', type=FileType('w+b'),
metavar='FILE',
help=SUPPRESS if not is_windows else _(
'''
Save output to FILE.
This option is a replacement for piping output to FILE,
which would on Windows result in corrupted data
being saved.
'''
)
)
output_processing.add_argument(
'--pretty', dest='prettify', default=PRETTY_STDOUT_TTY_ONLY,
'--pretty',
dest='prettify',
default=PRETTY_STDOUT_TTY_ONLY,
choices=sorted(PRETTY_MAP.keys()),
help=_('''
Controls output processing. The value can be "none" to not prettify
@ -140,7 +131,10 @@ output_processing.add_argument(
''')
)
output_processing.add_argument(
'--style', '-s', dest='style', default=DEFAULT_STYLE, metavar='STYLE',
'--style', '-s',
dest='style',
metavar='STYLE',
default=DEFAULT_STYLE,
choices=AVAILABLE_STYLES,
help=_('''
Output coloring style. One of %s. Defaults to "%s".
@ -157,7 +151,9 @@ output_processing.add_argument(
output_options = parser.add_argument_group(title='Output options')
output_options.add_argument(
'--print', '-p', dest='output_options', metavar='WHAT',
'--print', '-p',
dest='output_options',
metavar='WHAT',
help=_('''
String specifying what the output should contain:
"{request_headers}" stands for the request headers, and
@ -174,24 +170,30 @@ output_options.add_argument(
response_body=OUT_RESP_BODY,))
)
output_options.add_argument(
'--verbose', '-v', dest='output_options',
action='store_const', const=''.join(OUTPUT_OPTIONS),
'--verbose', '-v',
dest='output_options',
action='store_const',
const=''.join(OUTPUT_OPTIONS),
help=_('''
Print the whole request as well as the response.
Shortcut for --print={0}.
'''.format(''.join(OUTPUT_OPTIONS)))
)
output_options.add_argument(
'--headers', '-h', dest='output_options',
action='store_const', const=OUT_RESP_HEAD,
'--headers', '-h',
dest='output_options',
action='store_const',
const=OUT_RESP_HEAD,
help=_('''
Print only the response headers.
Shortcut for --print={0}.
'''.format(OUT_RESP_HEAD))
)
output_options.add_argument(
'--body', '-b', dest='output_options',
action='store_const', const=OUT_RESP_BODY,
'--body', '-b',
dest='output_options',
action='store_const',
const=OUT_RESP_BODY,
help=_('''
Print only the response body.
Shortcut for --print={0}.
@ -199,7 +201,9 @@ output_options.add_argument(
)
output_options.add_argument(
'--stream', '-S', action='store_true', default=False,
'--stream', '-S',
action='store_true',
default=False,
help=_('''
Always stream the output by line, i.e., behave like `tail -f'.
@ -214,7 +218,43 @@ output_options.add_argument(
''')
)
output_processing.add_argument(
'--output', '-o',
type=FileType('a+b'),
dest='output_file',
metavar='FILE',
help=_(
'''
Save output to FILE. If --download is set, then only the response
body is saved to the file. Other parts of the HTTP exchange are
printed to stderr.
'''
)
)
output_options.add_argument(
'--download', '-d',
action='store_true',
default=False,
help=_('''
Do not print the response body to stdout. Rather, download it and store it
in a file. The filename is guessed unless specified with --output
[filename]. This action is similar to the default behaviour of wget.
''')
)
output_options.add_argument(
'--continue', '-c',
dest='download_resume',
action='store_true',
default=False,
help=_('''
Resume an interrupted download.
The --output option needs to be specified as well.
''')
)
###############################################################################
# Sessions
@ -223,7 +263,9 @@ sessions = parser.add_argument_group(title='Sessions')\
.add_mutually_exclusive_group(required=False)
sessions.add_argument(
'--session', metavar='SESSION_NAME', type=RegexValidator(
'--session',
metavar='SESSION_NAME',
type=RegexValidator(
Session.VALID_NAME_PATTERN,
'Session name contains invalid characters.'
),
@ -235,7 +277,8 @@ sessions.add_argument(
''' % DEFAULT_SESSIONS_DIR)
)
sessions.add_argument(
'--session-read-only', metavar='SESSION_NAME',
'--session-read-only',
metavar='SESSION_NAME',
help=_('''
Create or read a session without updating it form the
request/response exchange.
@ -249,7 +292,8 @@ sessions.add_argument(
# ``requests.request`` keyword arguments.
auth = parser.add_argument_group(title='Authentication')
auth.add_argument(
'--auth', '-a', metavar='USER[:PASS]',
'--auth', '-a',
metavar='USER[:PASS]',
type=AuthCredentialsArgType(SEP_CREDENTIALS),
help=_('''
If only the username is provided (-a username),
@ -258,7 +302,9 @@ auth.add_argument(
)
auth.add_argument(
'--auth-type', choices=['basic', 'digest'], default='basic',
'--auth-type',
choices=['basic', 'digest'],
default='basic',
help=_('''
The authentication mechanism to be used.
Defaults to "basic".
@ -272,7 +318,10 @@ auth.add_argument(
network = parser.add_argument_group(title='Network')
network.add_argument(
'--proxy', default=[], action='append', metavar='PROTOCOL:HOST',
'--proxy',
default=[],
action='append',
metavar='PROTOCOL:HOST',
type=KeyValueArgType(SEP_PROXY),
help=_('''
String mapping protocol to the URL of the proxy
@ -281,14 +330,17 @@ network.add_argument(
''')
)
network.add_argument(
'--follow', default=False, action='store_true',
'--follow',
default=False,
action='store_true',
help=_('''
Set this flag if full redirects are allowed
(e.g. re-POST-ing of data at new ``Location``)
''')
)
network.add_argument(
'--verify', default='yes',
'--verify',
default='yes',
help=_('''
Set to "no" to skip checking the host\'s SSL certificate.
You can also pass the path to a CA_BUNDLE
@ -299,14 +351,19 @@ network.add_argument(
)
network.add_argument(
'--timeout', type=float, default=30, metavar='SECONDS',
'--timeout',
type=float,
default=30,
metavar='SECONDS',
help=_('''
The connection timeout of the request in seconds.
The default value is 30 seconds.
''')
)
network.add_argument(
'--check-status', default=False, action='store_true',
'--check-status',
default=False,
action='store_true',
help=_('''
By default, HTTPie exits with 0 when no network or other fatal
errors occur.
@ -333,17 +390,25 @@ troubleshooting = parser.add_argument_group(title='Troubleshooting')
troubleshooting.add_argument(
'--help',
action='help', default=SUPPRESS,
action='help',
default=SUPPRESS,
help='Show this help message and exit'
)
troubleshooting.add_argument(
'--version', action='version', version=__version__)
'--version',
action='version',
version=__version__
)
troubleshooting.add_argument(
'--traceback', action='store_true', default=False,
'--traceback',
action='store_true',
default=False,
help='Prints exception traceback should one occur.'
)
troubleshooting.add_argument(
'--debug', action='store_true', default=False,
'--debug',
action='store_true',
default=False,
help=_('''
Prints exception traceback should one occur, and also other
information that is useful for debugging HTTPie itself and

View File

@ -21,8 +21,9 @@ from pygments import __version__ as pygments_version
from .cli import parser
from .compat import str, is_py3
from .client import get_response
from .downloads import Download
from .models import Environment
from .output import build_output_stream, write, write_with_colors_win_p3k
from .output import build_output_stream, write, write_with_colors_win_py3
from . import ExitStatus
@ -77,6 +78,16 @@ def main(args=sys.argv[1:], env=Environment()):
try:
args = parser.parse_args(args=args, env=env)
download = None
if args.download:
args.follow = True
download = Download(
output_file=args.output_file,
progress_file=env.stderr,
resume=args.download_resume
)
download.alter_request_headers(args.headers)
response = get_response(args, config_dir=env.config.directory)
if args.check_status:
@ -89,18 +100,31 @@ def main(args=sys.argv[1:], env=Environment()):
level='warning')
write_kwargs = {
'stream': build_output_stream(args, env,
response.request,
response),
'stream': build_output_stream(
args, env, response.request, response),
# This will in fact be `stderr` with `--download`
'outfile': env.stdout,
'flush': env.stdout_isatty or args.stream
}
try:
if env.is_windows and is_py3 and 'colors' in args.prettify:
write_with_colors_win_p3k(**write_kwargs)
write_with_colors_win_py3(**write_kwargs)
else:
write(**write_kwargs)
if download:
# Response body download.
download_stream, download_to = download.start(response)
write(
stream=download_stream,
outfile=download_to,
flush=False,
)
except IOError as e:
if not traceback and e.errno == errno.EPIPE:
# Ignore broken pipes unless --traceback.

194
httpie/downloads.py Normal file
View File

@ -0,0 +1,194 @@
"""
Download mode implementation.
"""
from __future__ import division
import mimetypes
import os
import sys
import errno
from time import time
from .output import RawStream
from .models import HTTPResponse
from .humanize import humanize_bytes
from .compat import urlsplit
CLEAR_LINE = '\r\033[K'
PROGRESS_TPL = '{percentage:0.2f} % ({downloaded}) of {total} ({speed}/s)'
FINISHED_TPL = '{downloaded} of {total} in {time}s ({speed}/s)'
class Download(object):
def __init__(self, output_file=None,
resume=False, progress_file=sys.stderr):
"""
:param resume: Should the download resume if partial download
already exists.
:type resume: bool
:param output_file: The file to store response body in. If not
provided, it will be guessed from the response.
:type output_file: file
:param progress_file: Where to report download progress.
:type progress_file: file
"""
self.output_file = output_file
self.progress_file = progress_file
self.resume = resume
self.bytes_resumed_from = 0
self.bytes_total = 0
self.bytes_downloaded = 0
self.bytes_downloaded_prev = 0
self.bytes_total_humanized = ''
self.time_started = None
self.time_prev = None
self.speed = 0
def alter_request_headers(self, headers):
"""Called just before a download request is sent."""
# Disable content encoding so that we can resume, etc.
headers['Accept-Encoding'] = ''
if self.resume:
try:
bytes_have = os.path.getsize(self.output_file.name)
except OSError as e:
if e.errno != errno.ENOENT:
raise
else:
self.bytes_resumed_from = self.bytes_downloaded = bytes_have
# Set ``Range`` header to resume the download
headers['Range'] = '%d-' % bytes_have
def start(self, response):
"""
Initiate and return a stream for `response` body with progress
callback attached. Can be called only once.
:param response: Initiated response object.
:type response: requests.models.Response
:return: RawStream, output_file
"""
assert not self.time_started
if self.output_file:
if not self.resume:
self.output_file.seek(0)
self.output_file.truncate()
else:
# TODO: should we take the filename from response.history[0].url?
# TODO: --download implies --follow
# Output file not specified. Pick a name that doesn't exist yet.
content_type = response.headers.get('Content-Type', '')
self.output_file = open(
self._get_unique_output_filename(
url=response.url,
content_type=content_type,
),
mode='a+b'
)
self.bytes_total = int(response.headers.get('Content-Length'), 0)
self.bytes_total_humanized = humanize_bytes(self.bytes_total)
self.time_started = time()
self.time_prev = self.time_started
stream = RawStream(
msg=HTTPResponse(response),
with_headers=False,
with_body=True,
on_body_chunk_downloaded=self._on_progress
)
self.progress_file.write('Saving to %s\n' % self.output_file.name)
self._report_status()
return stream, self.output_file
def has_finished(self):
return self.bytes_downloaded == self.bytes_total
def _get_unique_output_filename(self, url, content_type):
suffix = 0
fn = None
while True:
fn = self._get_output_filename(
url=url,
content_type=content_type,
suffix=suffix
)
if not os.path.exists(fn):
break
suffix += 1
return fn
def _get_output_filename(self, url, content_type, suffix=None):
fn = urlsplit(url).path.rstrip('/')
fn = os.path.basename(fn) if fn else 'index'
if suffix:
fn += '-' + str(suffix)
if '.' not in fn:
ext = mimetypes.guess_extension(content_type.split(';')[0])
if ext:
fn += ext
return fn
def _on_progress(self, chunk):
"""
A download progress callback.
:param chunk: A chunk of response body data that has just
been downloaded and written to the output.
:type chunk: bytes
"""
self.bytes_downloaded += len(chunk)
self._report_status()
def _report_status(self):
now = time()
# Update the reported speed on the first chunk and once in a while.
if not self.bytes_downloaded_prev or now - self.time_prev >= .6:
self.speed = (
(self.bytes_downloaded - self.bytes_downloaded_prev)
/ (now - self.time_prev)
)
self.time_prev = now
self.bytes_downloaded_prev = self.bytes_downloaded
self.progress_file.write(CLEAR_LINE)
self.progress_file.write(PROGRESS_TPL.format(
percentage=self.bytes_downloaded / self.bytes_total * 100,
downloaded=humanize_bytes(self.bytes_downloaded),
total=self.bytes_total_humanized,
speed=humanize_bytes(self.speed)
))
# Report avg. speed and total time when finished.
if self.has_finished():
bytes_downloaded = self.bytes_downloaded - self.bytes_resumed_from
time_taken = time() - self.time_started
self.progress_file.write(CLEAR_LINE)
self.progress_file.write(FINISHED_TPL.format(
downloaded=humanize_bytes(bytes_downloaded),
total=humanize_bytes(self.bytes_total),
speed=humanize_bytes(bytes_downloaded / time_taken),
time=time_taken,
))
self.progress_file.write('\n')
self.output_file.close()
self.progress_file.flush()

53
httpie/humanize.py Normal file
View File

@ -0,0 +1,53 @@
"""
Author: Doug Latornell
Licence: MIT
URL: http://code.activestate.com/recipes/577081/
"""
import doctest
def humanize_bytes(n, precision=1):
"""Return a humanized string representation of a number of bytes.
Assumes `from __future__ import division`.
>>> humanize_bytes(1)
'1 byte'
>>> humanize_bytes(1024)
'1.0 kB'
>>> humanize_bytes(1024 * 123)
'123.0 kB'
>>> humanize_bytes(1024 * 12342)
'12.1 MB'
>>> humanize_bytes(1024 * 12342, 2)
'12.05 MB'
>>> humanize_bytes(1024 * 1234, 2)
'1.21 MB'
>>> humanize_bytes(1024 * 1234 * 1111, 2)
'1.31 GB'
>>> humanize_bytes(1024 * 1234 * 1111, 1)
'1.3 GB'
"""
abbrevs = [
(1 << 50, 'PB'),
(1 << 40, 'TB'),
(1 << 30, 'GB'),
(1 << 20, 'MB'),
(1 << 10, 'kB'),
(1, 'bytes')
]
if n == 1:
return '1 byte'
for factor, suffix in abbrevs:
if n >= factor:
break
return '%.*f %s' % (precision, n / factor, suffix)
if __name__ == '__main__':
doctest.testmod()

View File

@ -98,57 +98,94 @@ class Parser(ArgumentParser):
def parse_args(self, env, args=None, namespace=None):
self.env = env
self.args, no_options = super(Parser, self)\
.parse_known_args(args, namespace)
args, no_options = super(Parser, self).parse_known_args(args,
namespace)
self._apply_no_options(args, no_options)
if not args.json and env.config.implicit_content_type == 'form':
args.form = True
if args.debug:
args.traceback = True
if args.output:
env.stdout = args.output
env.stdout_isatty = False
self._process_output_options(args, env)
self._process_pretty_options(args, env)
self._guess_method(args, env)
self._parse_items(args)
if self.args.debug:
self.args.traceback = True
# Arguments processing and environment setup.
self._apply_no_options(no_options)
self._apply_config()
self._setup_standard_streams()
self._validate_download_options()
self._process_output_options()
self._process_pretty_options()
self._guess_method()
self._parse_items()
if not env.stdin_isatty:
self._body_from_file(args, env.stdin)
self._body_from_file(self.env.stdin)
if not (self.args.url.startswith(HTTP)
or self.args.url.startswith(HTTPS)):
# Default to 'https://' if invoked as `https args`.
scheme = HTTPS if self.env.progname == 'https' else HTTP
self.args.url = scheme + self.args.url
self._process_auth()
if not (args.url.startswith(HTTP) or args.url.startswith(HTTPS)):
scheme = HTTPS if env.progname == 'https' else HTTP
args.url = scheme + args.url
return self.args
self._process_auth(args)
# noinspection PyShadowingBuiltins
def _print_message(self, message, file=None):
# Sneak in our stderr/stdout.
file = {
sys.stdout: self.env.stdout,
sys.stderr: self.env.stderr,
None: self.env.stderr
}.get(file, file)
return args
super(Parser, self)._print_message(message, file)
def _process_auth(self, args):
url = urlsplit(args.url)
def _setup_standard_streams(self):
"""
Modify `env.stdout` and `env.stdout_isatty` based on args, if needed.
if args.auth:
if not args.auth.has_password():
"""
if self.args.download:
# With `--download`, we write everything that would normally go to
# `stdout` to `stderr` instead. Let's replace the stream so that
# we don't have to use many `if`s throughout the codebase.
# The response body will be treated separately.
self.env.stdout = self.env.stderr
self.env.stdout_isatty = self.env.stderr_isatty
elif self.args.output_file:
# When not `--download`ing, then `--output` simply replaces
# `stdout`. The file is opened for appending, which isn't what
# we want in this case.
self.args.output_file.seek(0)
self.args.output_file.truncate()
self.env.stdout = self.args.output_file
self.env.stdout_isatty = False
def _apply_config(self):
if (not self.args.json
and self.env.config.implicit_content_type == 'form'):
self.args.form = True
def _process_auth(self):
"""
If only a username provided via --auth, then ask for a password.
Or, take credentials from the URL, if provided.
"""
url = urlsplit(self.args.url)
if self.args.auth:
if not self.args.auth.has_password():
# Stdin already read (if not a tty) so it's save to prompt.
args.auth.prompt_password(url.netloc)
self.args.auth.prompt_password(url.netloc)
elif url.username is not None:
# Handle http://username:password@hostname/
username, password = url.username, url.password
args.auth = AuthCredentials(
self.args.auth = AuthCredentials(
key=username,
value=password,
sep=SEP_CREDENTIALS,
orig=SEP_CREDENTIALS.join([username, password])
)
def _apply_no_options(self, args, no_options):
def _apply_no_options(self, no_options):
"""For every `--no-OPTION` in `no_options`, set `args.OPTION` to
its default value. This allows for un-setting of options, e.g.,
specified in config.
@ -165,7 +202,7 @@ class Parser(ArgumentParser):
inverted = '--' + option[5:]
for action in self._actions:
if inverted in action.option_strings:
setattr(args, action.dest, action.default)
setattr(self.args, action.dest, action.default)
break
else:
invalid.append(option)
@ -174,123 +211,140 @@ class Parser(ArgumentParser):
msg = 'unrecognized arguments: %s'
self.error(msg % ' '.join(invalid))
def _print_message(self, message, file=None):
# Sneak in our stderr/stdout.
file = {
sys.stdout: self.env.stdout,
sys.stderr: self.env.stderr,
None: self.env.stderr
}.get(file, file)
super(Parser, self)._print_message(message, file)
def _body_from_file(self, args, fd):
def _body_from_file(self, fd):
"""There can only be one source of request data.
Bytes are always read.
"""
if args.data:
if self.args.data:
self.error('Request body (from stdin or a file) and request '
'data (key=value) cannot be mixed.')
args.data = getattr(fd, 'buffer', fd).read()
self.args.data = getattr(fd, 'buffer', fd).read()
def _guess_method(self, args, env):
def _guess_method(self):
"""Set `args.method` if not specified to either POST or GET
based on whether the request has data or not.
"""
if args.method is None:
if self.args.method is None:
# Invoked as `http URL'.
assert not args.items
if not env.stdin_isatty:
args.method = HTTP_POST
assert not self.args.items
if not self.env.stdin_isatty:
self.args.method = HTTP_POST
else:
args.method = HTTP_GET
self.args.method = HTTP_GET
# FIXME: False positive, e.g., "localhost" matches but is a valid URL.
elif not re.match('^[a-zA-Z]+$', args.method):
elif not re.match('^[a-zA-Z]+$', self.args.method):
# Invoked as `http URL item+'. The URL is now in `args.method`
# and the first ITEM is now incorrectly in `args.url`.
try:
# Parse the URL as an ITEM and store it as the first ITEM arg.
args.items.insert(
0, KeyValueArgType(*SEP_GROUP_ITEMS).__call__(args.url))
self.args.items.insert(
0,
KeyValueArgType(*SEP_GROUP_ITEMS).__call__(self.args.url)
)
except ArgumentTypeError as e:
if args.traceback:
if self.args.traceback:
raise
self.error(e.message)
else:
# Set the URL correctly
args.url = args.method
self.args.url = self.args.method
# Infer the method
has_data = not env.stdin_isatty or any(
item.sep in SEP_GROUP_DATA_ITEMS for item in args.items)
args.method = HTTP_POST if has_data else HTTP_GET
has_data = not self.env.stdin_isatty or any(
item.sep in SEP_GROUP_DATA_ITEMS
for item in self.args.items
)
self.args.method = HTTP_POST if has_data else HTTP_GET
def _parse_items(self, args):
"""Parse `args.items` into `args.headers`, `args.data`,
`args.`, and `args.files`.
def _parse_items(self):
"""Parse `args.items` into `args.headers`, `args.data`, `args.params`,
and `args.files`.
"""
args.headers = CaseInsensitiveDict()
args.data = ParamDict() if args.form else OrderedDict()
args.files = OrderedDict()
args.params = ParamDict()
self.args.headers = CaseInsensitiveDict()
self.args.data = ParamDict() if self.args.form else OrderedDict()
self.args.files = OrderedDict()
self.args.params = ParamDict()
try:
parse_items(items=args.items,
headers=args.headers,
data=args.data,
files=args.files,
params=args.params)
parse_items(items=self.args.items,
headers=self.args.headers,
data=self.args.data,
files=self.args.files,
params=self.args.params)
except ParseError as e:
if args.traceback:
if self.args.traceback:
raise
self.error(e.message)
if args.files and not args.form:
if self.args.files and not self.args.form:
# `http url @/path/to/file`
file_fields = list(args.files.keys())
file_fields = list(self.args.files.keys())
if file_fields != ['']:
self.error(
'Invalid file fields (perhaps you meant --form?): %s'
% ','.join(file_fields))
fn, fd = args.files['']
args.files = {}
self._body_from_file(args, fd)
if 'Content-Type' not in args.headers:
fn, fd = self.args.files['']
self.args.files = {}
self._body_from_file(fd)
if 'Content-Type' not in self.args.headers:
mime, encoding = mimetypes.guess_type(fn, strict=False)
if mime:
content_type = mime
if encoding:
content_type = '%s; charset=%s' % (mime, encoding)
args.headers['Content-Type'] = content_type
self.args.headers['Content-Type'] = content_type
def _process_output_options(self, args, env):
"""Apply defaults to output options or validate the provided ones.
def _process_output_options(self):
"""Apply defaults to output options, or validate the provided ones.
The default output options are stdout-type-sensitive.
"""
if not args.output_options:
args.output_options = (OUTPUT_OPTIONS_DEFAULT if env.stdout_isatty
else OUTPUT_OPTIONS_DEFAULT_STDOUT_REDIRECTED)
if not self.args.output_options:
self.args.output_options = (
OUTPUT_OPTIONS_DEFAULT
if self.env.stdout_isatty
else OUTPUT_OPTIONS_DEFAULT_STDOUT_REDIRECTED
)
unknown = set(args.output_options) - OUTPUT_OPTIONS
if unknown:
self.error('Unknown output options: %s' % ','.join(unknown))
unknown_output_options = set(self.args.output_options) - OUTPUT_OPTIONS
if unknown_output_options:
self.error(
'Unknown output options: %s' % ','.join(unknown_output_options)
)
def _process_pretty_options(self, args, env):
if args.prettify == PRETTY_STDOUT_TTY_ONLY:
args.prettify = PRETTY_MAP['all' if env.stdout_isatty else 'none']
elif args.prettify and env.is_windows:
if self.args.download and OUT_RESP_BODY in self.args.output_options:
# Response body is always downloaded with --download and it goes
# through a different routine, so we remove it.
self.args.output_options = str(
set(self.args.output_options) - set(OUT_RESP_BODY))
def _process_pretty_options(self):
if self.args.prettify == PRETTY_STDOUT_TTY_ONLY:
self.args.prettify = PRETTY_MAP[
'all' if self.env.stdout_isatty else 'none']
elif self.args.prettify and self.env.is_windows:
self.error('Only terminal output can be colorized on Windows.')
else:
args.prettify = PRETTY_MAP[args.prettify]
# noinspection PyTypeChecker
self.args.prettify = PRETTY_MAP[self.args.prettify]
def _validate_download_options(self):
if not self.args.download:
if self.args.download_resume:
self.error('--continue only works with --download')
if self.args.download_resume and not (
self.args.download and self.args.output_file):
self.error('--continue requires --output to be specified')
class ParseError(Exception):
@ -320,16 +374,6 @@ def session_name_arg_type(name):
return name
def host_name_arg_type(name):
from .sessions import Host
if not Host.is_valid_name(name):
raise ArgumentTypeError(
'special characters and spaces are not'
' allowed in host names: "%s"'
% name)
return name
class RegexValidator(object):
def __init__(self, pattern, error_message):

View File

@ -19,22 +19,25 @@ class Environment(object):
if progname not in ['http', 'https']:
progname = 'http'
stdin_isatty = sys.stdin.isatty()
stdin = sys.stdin
stdout_isatty = sys.stdout.isatty()
config_dir = DEFAULT_CONFIG_DIR
# Can be set to 0 to disable colors completely.
colors = 256 if '256color' in os.environ.get('TERM', '') else 88
stdin = sys.stdin
stdin_isatty = sys.stdin.isatty()
stdout_isatty = sys.stdout.isatty()
if stdout_isatty and is_windows:
# noinspection PyUnresolvedReferences
from colorama.initialise import wrap_stream
stdout = wrap_stream(sys.stdout, convert=None,
strip=None, autoreset=True, wrap=True)
else:
stdout = sys.stdout
stderr = sys.stderr
# Can be set to 0 to disable colors completely.
colors = 256 if '256color' in os.environ.get('TERM', '') else 88
stderr = sys.stderr
stderr_isatty = sys.stderr.isatty()
def __init__(self, **kwargs):
assert all(hasattr(type(self), attr)

View File

@ -61,7 +61,7 @@ def write(stream, outfile, flush):
outfile.flush()
def write_with_colors_win_p3k(stream, outfile, flush):
def write_with_colors_win_py3(stream, outfile, flush):
"""Like `write`, but colorized chunks are written as text
directly to `outfile` to ensure it gets processed by colorama.
Applies only to Windows with Python 3 and colorized terminal output.
@ -147,7 +147,8 @@ def get_stream_type(env, args):
class BaseStream(object):
"""Base HTTP message output stream class."""
def __init__(self, msg, with_headers=True, with_body=True):
def __init__(self, msg, with_headers=True, with_body=True,
on_body_chunk_downloaded=None):
"""
:param msg: a :class:`models.HTTPMessage` subclass
:param with_headers: if `True`, headers will be included
@ -158,6 +159,7 @@ class BaseStream(object):
self.msg = msg
self.with_headers = with_headers
self.with_body = with_body
self.on_body_chunk_downloaded = on_body_chunk_downloaded
def _get_headers(self):
"""Return the headers' bytes."""
@ -177,6 +179,8 @@ class BaseStream(object):
try:
for chunk in self._iter_body():
yield chunk
if self.on_body_chunk_downloaded:
self.on_body_chunk_downloaded(chunk)
except BinarySuppressedError as e:
if self.with_headers:
yield b'\n'

View File

@ -1 +1,2 @@
#
git://github.com/kennethreitz/httpbin.git

View File

@ -268,7 +268,7 @@ class HTTPieTest(BaseTestCase):
'foo=bar'
)
self.assertIn(OK, r)
self.assertIn('"foo": "bar"', r)
self.assertIn(r'\"foo\": \"bar\"', r)
def test_POST_JSON_data(self):
r = http(
@ -277,7 +277,7 @@ class HTTPieTest(BaseTestCase):
'foo=bar'
)
self.assertIn(OK, r)
self.assertIn('"foo": "bar"', r)
self.assertIn(r'\"foo\": \"bar\"', r)
def test_POST_form(self):
r = http(
@ -519,7 +519,7 @@ class ImplicitHTTPMethodTest(BaseTestCase):
'hello=world'
)
self.assertIn(OK, r)
self.assertIn('"hello": "world"', r)
self.assertIn(r'\"hello\": \"world\"', r)
def test_implicit_POST_form(self):
r = http(
@ -658,8 +658,8 @@ class VerboseFlagTest(BaseTestCase):
'baz=bar'
)
self.assertIn(OK, r)
#noinspection PyUnresolvedReferences
self.assertEqual(r.count('"baz": "bar"'), 2)
self.assertIn('"baz": "bar"', r) # request
self.assertIn(r'\"baz\": \"bar\"', r) # response
class MultipartFormDataFileUploadTest(BaseTestCase):
@ -845,8 +845,8 @@ class AuthTest(BaseTestCase):
httpbin('/digest-auth/auth/user/password')
)
self.assertIn(OK, r)
self.assertIn('"authenticated": true', r)
self.assertIn('"user": "user"', r)
self.assertIn(r'"authenticated": true', r)
self.assertIn(r'"user": "user"', r)
def test_password_prompt(self):
@ -1195,71 +1195,79 @@ class ArgumentParserTestCase(unittest.TestCase):
self.parser = input.Parser()
def test_guess_when_method_set_and_valid(self):
args = argparse.Namespace()
args.method = 'GET'
args.url = 'http://example.com/'
args.items = []
self.parser.args = argparse.Namespace()
self.parser.args.method = 'GET'
self.parser.args.url = 'http://example.com/'
self.parser.args.items = []
self.parser._guess_method(args, TestEnvironment())
self.parser.env = TestEnvironment()
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
self.assertEqual(args.items, [])
self.parser._guess_method()
self.assertEqual(self.parser.args.method, 'GET')
self.assertEqual(self.parser.args.url, 'http://example.com/')
self.assertEqual(self.parser.args.items, [])
def test_guess_when_method_not_set(self):
args = argparse.Namespace()
args.method = None
args.url = 'http://example.com/'
args.items = []
self.parser._guess_method(args, TestEnvironment())
self.parser.args = argparse.Namespace()
self.parser.args.method = None
self.parser.args.url = 'http://example.com/'
self.parser.args.items = []
self.parser.env = TestEnvironment()
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
self.assertEqual(args.items, [])
self.parser._guess_method()
self.assertEqual(self.parser.args.method, 'GET')
self.assertEqual(self.parser.args.url, 'http://example.com/')
self.assertEqual(self.parser.args.items, [])
def test_guess_when_method_set_but_invalid_and_data_field(self):
args = argparse.Namespace()
args.method = 'http://example.com/'
args.url = 'data=field'
args.items = []
self.parser.args = argparse.Namespace()
self.parser.args.method = 'http://example.com/'
self.parser.args.url = 'data=field'
self.parser.args.items = []
self.parser.env = TestEnvironment()
self.parser._guess_method()
self.parser._guess_method(args, TestEnvironment())
self.assertEqual(args.method, 'POST')
self.assertEqual(args.url, 'http://example.com/')
self.assertEqual(self.parser.args.method, 'POST')
self.assertEqual(self.parser.args.url, 'http://example.com/')
self.assertEqual(
args.items,
self.parser.args.items,
[input.KeyValue(
key='data', value='field', sep='=', orig='data=field')])
def test_guess_when_method_set_but_invalid_and_header_field(self):
args = argparse.Namespace()
args.method = 'http://example.com/'
args.url = 'test:header'
args.items = []
self.parser.args = argparse.Namespace()
self.parser.args.method = 'http://example.com/'
self.parser.args.url = 'test:header'
self.parser.args.items = []
self.parser._guess_method(args, TestEnvironment())
self.parser.env = TestEnvironment()
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
self.parser._guess_method()
self.assertEqual(self.parser.args.method, 'GET')
self.assertEqual(self.parser.args.url, 'http://example.com/')
self.assertEqual(
args.items,
self.parser.args.items,
[input.KeyValue(
key='test', value='header', sep=':', orig='test:header')])
def test_guess_when_method_set_but_invalid_and_item_exists(self):
args = argparse.Namespace()
args.method = 'http://example.com/'
args.url = 'new_item=a'
args.items = [
self.parser.args = argparse.Namespace()
self.parser.args.method = 'http://example.com/'
self.parser.args.url = 'new_item=a'
self.parser.args.items = [
input.KeyValue(
key='old_item', value='b', sep='=', orig='old_item=b')
]
self.parser._guess_method(args, TestEnvironment())
self.parser.env = TestEnvironment()
self.assertEqual(args.items, [
self.parser._guess_method()
self.assertEqual(self.parser.args.items, [
input.KeyValue(
key='new_item', value='a', sep='=', orig='new_item=a'),
input.KeyValue(
@ -1410,5 +1418,10 @@ class SessionTest(BaseTestCase):
self.assertDictEqual(r1.json, r3.json)
class DownloadsTest(BaseTestCase):
# TODO: tests for downloads
pass
if __name__ == '__main__':
unittest.main()

View File

@ -11,9 +11,3 @@ commands = {envpython} setup.py test
[testenv:py26]
deps = argparse
[testenv:py30]
deps = argparse
[testenv:py31]
deps = argparse