Method to check for "port is open" based on the URL

[jeremyschulman]

As a developer of client libraries that use HTTPX, specifically in asyncio, I find myself writing a function that will check to see if the port is open before attempting a number of API calls. For example if my default timeout is 60 seconds, I do not want the first API call to wait 60s just to tell me that the port is unavailable. As result, I have this code in each of my client libraries.

My feature request would be to add this method to the httpx.AsyncClient.

Thank you for your considerations.

Cheers,
-- Jeremy

# -----------------------------------------------------------------------------
# System Imports
# -----------------------------------------------------------------------------

from typing import Optional
import socket
import asyncio

# -----------------------------------------------------------------------------
# Public Imports
# -----------------------------------------------------------------------------


from httpx import URL

# -----------------------------------------------------------------------------
# Exports
# -----------------------------------------------------------------------------

__all__ = ["port_check_url"]

# -----------------------------------------------------------------------------
#
#                                 CODE BEGINS
#
# -----------------------------------------------------------------------------


async def port_check_url(url: URL, timeout: Optional[int] = 5) -> bool:
    """
    This function attempts to open the port designated by the URL given the
    timeout in seconds.  If the port is avaialble then return True; False
    otherwise.

    Parameters
    ----------
    url:
        The URL that provides the target system

    timeout: optional, default is 5 seonds
        Time to await for the port to open in seconds
    """
    port = url.port or socket.getservbyname(url.scheme)

    try:
        wr: asyncio.StreamWriter
        _, wr = await asyncio.wait_for(
            asyncio.open_connection(host=url.host, port=port), timeout=timeout
        )

        # MUST close if opened!
        wr.close()
        return True

    except Exception:  # noqa
        return False

[ghost]

Can't you set a timeout using the timeout= argument?

import httpx

default_timeout = 40 # for everything else
connect_timeout = 5 # for connection

httpx.get("http://google.com:1337", timeout=httpx.Timeout(default_timeout, connect=connect_timeout))

See the documentation for more information