1  from __future__ import annotations
   2  
   3  import functools
   4  import json
   5  import sys
   6  import typing
   7  
   8  import click
   9  import pygments.lexers
  10  import pygments.util
  11  import rich.console
  12  import rich.markup
  13  import rich.progress
  14  import rich.syntax
  15  import rich.table
  16  
  17  from ._client import Client
  18  from ._exceptions import RequestError
  19  from ._models import Response
  20  from ._status_codes import codes
  21  
  22  if typing.TYPE_CHECKING:
  23      import httpcore  # pragma: no cover
  24  
  25  
  26  def print_help() -> None:
  27      console = rich.console.Console()
  28  
  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()
  37  
  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      )
  75  
  76      table.add_row(
  77          "--proxy [cyan]URL",
  78          "Send the request via a proxy. Should be the URL giving the proxy address.",
  79      )
  80  
  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      )
  86  
  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      )
  92  
  93      table.add_row(
  94          "--download [cyan]FILE",
  95          "Save the response content as a file, rather than displaying it.",
  96      )
  97  
  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)
 101  
 102  
 103  def 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
 114  
 115  
 116  def 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)
 127  
 128  
 129  def 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)
 145  
 146  
 147  def 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)
 154  
 155  
 156  def 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)
 168  
 169  
 170  def 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
 182  
 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>")
 187  
 188  
 189  _PCTRTT = typing.Tuple[typing.Tuple[str, str], ...]
 190  _PCTRTTT = typing.Tuple[_PCTRTT, ...]
 191  _PeerCertRetDictType = typing.Dict[str, typing.Union[str, _PCTRTTT, _PCTRTT]]
 192  
 193  
 194  def 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)
 210  
 211  
 212  def 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)
 249  
 250  
 251  def 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)
 271  
 272  
 273  def 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
 280  
 281      try:
 282          return json.loads(value)
 283      except json.JSONDecodeError:  # pragma: no cover
 284          raise click.BadParameter("Not valid JSON")
 285  
 286  
 287  def 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
 294  
 295      username, password = value
 296      if password == "-":  # pragma: no cover
 297          password = click.prompt("Password", hide_input=True)
 298      return (username, password)
 299  
 300  
 301  def 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
 308  
 309      print_help()
 310      ctx.exit()
 311  
 312  
 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  )
 452  def 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"
 477  
 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)
 500  
 501      except RequestError as exc:
 502          console = rich.console.Console()
 503          console.print(f"[red]{type(exc).__name__}[/red]: {exc}")
 504          sys.exit(1)
 505  
 506      sys.exit(0 if response.is_success else 1)
 507