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