codekingpro/portable-devtools

1from __future__ import annotations2 3import functools4import json5import sys6import typing7 8import click9import pygments.lexers10import pygments.util11import rich.console12import rich.markup13import rich.progress14import rich.syntax15import rich.table16 17from ._client import Client18from ._exceptions import RequestError19from ._models import Response20from ._status_codes import codes21 22if typing.TYPE_CHECKING:23    import httpcore  # pragma: no cover24 25 26def 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 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()).name110            )111        except pygments.util.ClassNotFound:  # pragma: no cover112            pass113    return ""  # pragma: no cover114 115 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.headers120    ]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 headers125    ]126    return "\n".join(lines)127 128 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 None139        else reason_phrase.decode("ascii")140    )141    lines = [f"{version} {status} {reason}"] + [142        f"{name.decode('ascii')}: {value.decode('ascii')}" for name, value in headers143    ]144    return "\n".join(lines)145 146 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)154 155 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)168 169 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 cover179                text = response.text180        else:181            text = response.text182 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 194def format_certificate(cert: _PeerCertRetDictType) -> str:  # pragma: no cover195    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 212def trace(213    name: str, info: typing.Mapping[str, typing.Any], verbose: bool = False214) -> 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 cover224        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 cover239        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 cover245        status, headers = info["return_value"]246        http_version = b"HTTP/2"247        reason_phrase = None248        print_response_headers(http_version, status, reason_phrase, headers)249 250 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)271 272 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 None280 281    try:282        return json.loads(value)283    except json.JSONDecodeError:  # pragma: no cover284        raise click.BadParameter("Not valid JSON")285 286 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 None294 295    username, password = value296    if password == "-":  # pragma: no cover297        password = click.prompt("Password", hide_input=True)298    return (username, password)299 300 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        return308 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)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"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: ignore487                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