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