Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpx/_main.py: 5%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import functools
4import json
5import sys
6import typing
8import click
9import httpcore
10import pygments.lexers
11import pygments.util
12import rich.console
13import rich.markup
14import rich.progress
15import rich.syntax
16import rich.table
18from ._client import Client
19from ._exceptions import RequestError
20from ._models import Response
21from ._status_codes import codes
24def print_help() -> None:
25 console = rich.console.Console()
27 console.print("[bold]HTTPX :butterfly:", justify="center")
28 console.print()
29 console.print("A next generation HTTP client.", justify="center")
30 console.print()
31 console.print(
32 "Usage: [bold]httpx[/bold] [cyan]<URL> [OPTIONS][/cyan] ", justify="left"
33 )
34 console.print()
36 table = rich.table.Table.grid(padding=1, pad_edge=True)
37 table.add_column("Parameter", no_wrap=True, justify="left", style="bold")
38 table.add_column("Description")
39 table.add_row(
40 "-m, --method [cyan]METHOD",
41 "Request method, such as GET, POST, PUT, PATCH, DELETE, OPTIONS, HEAD.\n"
42 "[Default: GET, or POST if a request body is included]",
43 )
44 table.add_row(
45 "-p, --params [cyan]<NAME VALUE> ...",
46 "Query parameters to include in the request URL.",
47 )
48 table.add_row(
49 "-c, --content [cyan]TEXT", "Byte content to include in the request body."
50 )
51 table.add_row(
52 "-d, --data [cyan]<NAME VALUE> ...", "Form data to include in the request body."
53 )
54 table.add_row(
55 "-f, --files [cyan]<NAME FILENAME> ...",
56 "Form files to include in the request body.",
57 )
58 table.add_row("-j, --json [cyan]TEXT", "JSON data to include in the request body.")
59 table.add_row(
60 "-h, --headers [cyan]<NAME VALUE> ...",
61 "Include additional HTTP headers in the request.",
62 )
63 table.add_row(
64 "--cookies [cyan]<NAME VALUE> ...", "Cookies to include in the request."
65 )
66 table.add_row(
67 "--auth [cyan]<USER PASS>",
68 "Username and password to include in the request. Specify '-' for the password"
69 " to use a password prompt. Note that using --verbose/-v will expose"
70 " the Authorization header, including the password encoding"
71 " in a trivially reversible format.",
72 )
74 table.add_row(
75 "--proxy [cyan]URL",
76 "Send the request via a proxy. Should be the URL giving the proxy address.",
77 )
79 table.add_row(
80 "--timeout [cyan]FLOAT",
81 "Timeout value to use for network operations, such as establishing the"
82 " connection, reading some data, etc... [Default: 5.0]",
83 )
85 table.add_row("--follow-redirects", "Automatically follow redirects.")
86 table.add_row("--no-verify", "Disable SSL verification.")
87 table.add_row(
88 "--http2", "Send the request using HTTP/2, if the remote server supports it."
89 )
91 table.add_row(
92 "--download [cyan]FILE",
93 "Save the response content as a file, rather than displaying it.",
94 )
96 table.add_row("-v, --verbose", "Verbose output. Show request as well as response.")
97 table.add_row("--help", "Show this message and exit.")
98 console.print(table)
101def get_lexer_for_response(response: Response) -> str:
102 content_type = response.headers.get("Content-Type")
103 if content_type is not None:
104 mime_type, _, _ = content_type.partition(";")
105 try:
106 return typing.cast(
107 str, pygments.lexers.get_lexer_for_mimetype(mime_type.strip()).name
108 )
109 except pygments.util.ClassNotFound: # pragma: no cover
110 pass
111 return "" # pragma: no cover
114def format_request_headers(request: httpcore.Request, http2: bool = False) -> str:
115 version = "HTTP/2" if http2 else "HTTP/1.1"
116 headers = [
117 (name.lower() if http2 else name, value) for name, value in request.headers
118 ]
119 method = request.method.decode("ascii")
120 target = request.url.target.decode("ascii")
121 lines = [f"{method} {target} {version}"] + [
122 f"{name.decode('ascii')}: {value.decode('ascii')}" for name, value in headers
123 ]
124 return "\n".join(lines)
127def format_response_headers(
128 http_version: bytes,
129 status: int,
130 reason_phrase: bytes | None,
131 headers: list[tuple[bytes, bytes]],
132) -> str:
133 version = http_version.decode("ascii")
134 reason = (
135 codes.get_reason_phrase(status)
136 if reason_phrase is None
137 else reason_phrase.decode("ascii")
138 )
139 lines = [f"{version} {status} {reason}"] + [
140 f"{name.decode('ascii')}: {value.decode('ascii')}" for name, value in headers
141 ]
142 return "\n".join(lines)
145def print_request_headers(request: httpcore.Request, http2: bool = False) -> None:
146 console = rich.console.Console()
147 http_text = format_request_headers(request, http2=http2)
148 syntax = rich.syntax.Syntax(http_text, "http", theme="ansi_dark", word_wrap=True)
149 console.print(syntax)
150 syntax = rich.syntax.Syntax("", "http", theme="ansi_dark", word_wrap=True)
151 console.print(syntax)
154def print_response_headers(
155 http_version: bytes,
156 status: int,
157 reason_phrase: bytes | None,
158 headers: list[tuple[bytes, bytes]],
159) -> None:
160 console = rich.console.Console()
161 http_text = format_response_headers(http_version, status, reason_phrase, headers)
162 syntax = rich.syntax.Syntax(http_text, "http", theme="ansi_dark", word_wrap=True)
163 console.print(syntax)
164 syntax = rich.syntax.Syntax("", "http", theme="ansi_dark", word_wrap=True)
165 console.print(syntax)
168def print_response(response: Response) -> None:
169 console = rich.console.Console()
170 lexer_name = get_lexer_for_response(response)
171 if lexer_name:
172 if lexer_name.lower() == "json":
173 try:
174 data = response.json()
175 text = json.dumps(data, indent=4)
176 except ValueError: # pragma: no cover
177 text = response.text
178 else:
179 text = response.text
181 syntax = rich.syntax.Syntax(text, lexer_name, theme="ansi_dark", word_wrap=True)
182 console.print(syntax)
183 else:
184 console.print(f"<{len(response.content)} bytes of binary data>")
187_PCTRTT = typing.Tuple[typing.Tuple[str, str], ...]
188_PCTRTTT = typing.Tuple[_PCTRTT, ...]
189_PeerCertRetDictType = typing.Dict[str, typing.Union[str, _PCTRTTT, _PCTRTT]]
192def format_certificate(cert: _PeerCertRetDictType) -> str: # pragma: no cover
193 lines = []
194 for key, value in cert.items():
195 if isinstance(value, (list, tuple)):
196 lines.append(f"* {key}:")
197 for item in value:
198 if key in ("subject", "issuer"):
199 for sub_item in item:
200 lines.append(f"* {sub_item[0]}: {sub_item[1]!r}")
201 elif isinstance(item, tuple) and len(item) == 2:
202 lines.append(f"* {item[0]}: {item[1]!r}")
203 else:
204 lines.append(f"* {item!r}")
205 else:
206 lines.append(f"* {key}: {value!r}")
207 return "\n".join(lines)
210def trace(
211 name: str, info: typing.Mapping[str, typing.Any], verbose: bool = False
212) -> None:
213 console = rich.console.Console()
214 if name == "connection.connect_tcp.started" and verbose:
215 host = info["host"]
216 console.print(f"* Connecting to {host!r}")
217 elif name == "connection.connect_tcp.complete" and verbose:
218 stream = info["return_value"]
219 server_addr = stream.get_extra_info("server_addr")
220 console.print(f"* Connected to {server_addr[0]!r} on port {server_addr[1]}")
221 elif name == "connection.start_tls.complete" and verbose: # pragma: no cover
222 stream = info["return_value"]
223 ssl_object = stream.get_extra_info("ssl_object")
224 version = ssl_object.version()
225 cipher = ssl_object.cipher()
226 server_cert = ssl_object.getpeercert()
227 alpn = ssl_object.selected_alpn_protocol()
228 console.print(f"* SSL established using {version!r} / {cipher[0]!r}")
229 console.print(f"* Selected ALPN protocol: {alpn!r}")
230 if server_cert:
231 console.print("* Server certificate:")
232 console.print(format_certificate(server_cert))
233 elif name == "http11.send_request_headers.started" and verbose:
234 request = info["request"]
235 print_request_headers(request, http2=False)
236 elif name == "http2.send_request_headers.started" and verbose: # pragma: no cover
237 request = info["request"]
238 print_request_headers(request, http2=True)
239 elif name == "http11.receive_response_headers.complete":
240 http_version, status, reason_phrase, headers = info["return_value"]
241 print_response_headers(http_version, status, reason_phrase, headers)
242 elif name == "http2.receive_response_headers.complete": # pragma: no cover
243 status, headers = info["return_value"]
244 http_version = b"HTTP/2"
245 reason_phrase = None
246 print_response_headers(http_version, status, reason_phrase, headers)
249def download_response(response: Response, download: typing.BinaryIO) -> None:
250 console = rich.console.Console()
251 console.print()
252 content_length = response.headers.get("Content-Length")
253 with rich.progress.Progress(
254 "[progress.description]{task.description}",
255 "[progress.percentage]{task.percentage:>3.0f}%",
256 rich.progress.BarColumn(bar_width=None),
257 rich.progress.DownloadColumn(),
258 rich.progress.TransferSpeedColumn(),
259 ) as progress:
260 description = f"Downloading [bold]{rich.markup.escape(download.name)}"
261 download_task = progress.add_task(
262 description,
263 total=int(content_length or 0),
264 start=content_length is not None,
265 )
266 for chunk in response.iter_bytes():
267 download.write(chunk)
268 progress.update(download_task, completed=response.num_bytes_downloaded)
271def validate_json(
272 ctx: click.Context,
273 param: click.Option | click.Parameter,
274 value: typing.Any,
275) -> typing.Any:
276 if value is None:
277 return None
279 try:
280 return json.loads(value)
281 except json.JSONDecodeError: # pragma: no cover
282 raise click.BadParameter("Not valid JSON")
285def validate_auth(
286 ctx: click.Context,
287 param: click.Option | click.Parameter,
288 value: typing.Any,
289) -> typing.Any:
290 if value == (None, None):
291 return None
293 username, password = value
294 if password == "-": # pragma: no cover
295 password = click.prompt("Password", hide_input=True)
296 return (username, password)
299def handle_help(
300 ctx: click.Context,
301 param: click.Option | click.Parameter,
302 value: typing.Any,
303) -> None:
304 if not value or ctx.resilient_parsing:
305 return
307 print_help()
308 ctx.exit()
311@click.command(add_help_option=False)
312@click.argument("url", type=str)
313@click.option(
314 "--method",
315 "-m",
316 "method",
317 type=str,
318 help=(
319 "Request method, such as GET, POST, PUT, PATCH, DELETE, OPTIONS, HEAD. "
320 "[Default: GET, or POST if a request body is included]"
321 ),
322)
323@click.option(
324 "--params",
325 "-p",
326 "params",
327 type=(str, str),
328 multiple=True,
329 help="Query parameters to include in the request URL.",
330)
331@click.option(
332 "--content",
333 "-c",
334 "content",
335 type=str,
336 help="Byte content to include in the request body.",
337)
338@click.option(
339 "--data",
340 "-d",
341 "data",
342 type=(str, str),
343 multiple=True,
344 help="Form data to include in the request body.",
345)
346@click.option(
347 "--files",
348 "-f",
349 "files",
350 type=(str, click.File(mode="rb")),
351 multiple=True,
352 help="Form files to include in the request body.",
353)
354@click.option(
355 "--json",
356 "-j",
357 "json",
358 type=str,
359 callback=validate_json,
360 help="JSON data to include in the request body.",
361)
362@click.option(
363 "--headers",
364 "-h",
365 "headers",
366 type=(str, str),
367 multiple=True,
368 help="Include additional HTTP headers in the request.",
369)
370@click.option(
371 "--cookies",
372 "cookies",
373 type=(str, str),
374 multiple=True,
375 help="Cookies to include in the request.",
376)
377@click.option(
378 "--auth",
379 "auth",
380 type=(str, str),
381 default=(None, None),
382 callback=validate_auth,
383 help=(
384 "Username and password to include in the request. "
385 "Specify '-' for the password to use a password prompt. "
386 "Note that using --verbose/-v will expose the Authorization header, "
387 "including the password encoding in a trivially reversible format."
388 ),
389)
390@click.option(
391 "--proxy",
392 "proxy",
393 type=str,
394 default=None,
395 help="Send the request via a proxy. Should be the URL giving the proxy address.",
396)
397@click.option(
398 "--timeout",
399 "timeout",
400 type=float,
401 default=5.0,
402 help=(
403 "Timeout value to use for network operations, such as establishing the "
404 "connection, reading some data, etc... [Default: 5.0]"
405 ),
406)
407@click.option(
408 "--follow-redirects",
409 "follow_redirects",
410 is_flag=True,
411 default=False,
412 help="Automatically follow redirects.",
413)
414@click.option(
415 "--no-verify",
416 "verify",
417 is_flag=True,
418 default=True,
419 help="Disable SSL verification.",
420)
421@click.option(
422 "--http2",
423 "http2",
424 type=bool,
425 is_flag=True,
426 default=False,
427 help="Send the request using HTTP/2, if the remote server supports it.",
428)
429@click.option(
430 "--download",
431 type=click.File("wb"),
432 help="Save the response content as a file, rather than displaying it.",
433)
434@click.option(
435 "--verbose",
436 "-v",
437 type=bool,
438 is_flag=True,
439 default=False,
440 help="Verbose. Show request as well as response.",
441)
442@click.option(
443 "--help",
444 is_flag=True,
445 is_eager=True,
446 expose_value=False,
447 callback=handle_help,
448 help="Show this message and exit.",
449)
450def main(
451 url: str,
452 method: str,
453 params: list[tuple[str, str]],
454 content: str,
455 data: list[tuple[str, str]],
456 files: list[tuple[str, click.File]],
457 json: str,
458 headers: list[tuple[str, str]],
459 cookies: list[tuple[str, str]],
460 auth: tuple[str, str] | None,
461 proxy: str,
462 timeout: float,
463 follow_redirects: bool,
464 verify: bool,
465 http2: bool,
466 download: typing.BinaryIO | None,
467 verbose: bool,
468) -> None:
469 """
470 An HTTP command line client.
471 Sends a request and displays the response.
472 """
473 if not method:
474 method = "POST" if content or data or files or json else "GET"
476 try:
477 with Client(
478 proxy=proxy,
479 timeout=timeout,
480 verify=verify,
481 http2=http2,
482 ) as client:
483 with client.stream(
484 method,
485 url,
486 params=list(params),
487 content=content,
488 data=dict(data),
489 files=files, # type: ignore
490 json=json,
491 headers=headers,
492 cookies=dict(cookies),
493 auth=auth,
494 follow_redirects=follow_redirects,
495 extensions={"trace": functools.partial(trace, verbose=verbose)},
496 ) as response:
497 if download is not None:
498 download_response(response, download)
499 else:
500 response.read()
501 if response.content:
502 print_response(response)
504 except RequestError as exc:
505 console = rich.console.Console()
506 console.print(f"[red]{type(exc).__name__}[/red]: {exc}")
507 sys.exit(1)
509 sys.exit(0 if response.is_success else 1)