Skip to content

claim_port(port)

Claim port on every address that localhost can mean on this machine.

To accept connections, call listen() on each of the claimed port's sockets. To release the port, close the claimed port, not its sockets.

Linux: The sockets are already listening when they're returned. Calling listen() on them is still safe.

macOS: The sockets don't listen until you call listen() on them.

Windows: The sockets don't listen until you call listen() on them. While the port is claimed, another program can still bind it on :: with a socket that accepts IPv4 and IPv6 at once.

Parameters:

Name Type Description Default
port int

Port to claim.

required

Returns:

Type Description
ClaimedPort

The claimed port, with a bound socket for each address.

Raises:

Type Description
PortTypeError

When port isn't an int.

PortOutOfRangeError

When port isn't a valid port number.

PortInUseError

When port is in use and cannot be claimed.

OSError

When port can't be claimed for any other reason.

Examples:

As a context manager:

with claim_port(8000) as port:
    # Bound to 127.0.0.1, and ::1 when there's IPv6
    for sock in port.sockets:
        sock.listen()

# Leaving the `with` block closed the port.

Without a context manager, you must close the port yourself:

port = claim_port(8000)

try:
    for sock in port.sockets:
        sock.listen()

finally:
    port.close()
Source code in src/notwork/claim_port.py
def claim_port(port: int) -> ClaimedPort:
    """
    Claim `port` on every address that `localhost` can mean on this machine.

    To accept connections, call `listen()` on each of the claimed port's sockets. To
    release the port, close the claimed port, not its sockets.

    **Linux:** The sockets are already listening when they're returned. Calling
    `listen()` on them is still safe.

    **macOS:** The sockets don't listen until you call `listen()` on them.

    **Windows:** The sockets don't listen until you call `listen()` on them. While the
    port is claimed, another program can still bind it on `::` with a socket that
    accepts IPv4 and IPv6 at once.

    Args:
        port: Port to claim.

    Returns:
        The claimed port, with a bound socket for each address.

    Raises:
        PortTypeError: When `port` isn't an `int`.
        PortOutOfRangeError: When `port` isn't a valid port number.
        PortInUseError: When `port` is in use and cannot be claimed.
        OSError: When `port` can't be claimed for any other reason.

    Examples:
        As a context manager:

        ```python
        with claim_port(8000) as port:
            # Bound to 127.0.0.1, and ::1 when there's IPv6
            for sock in port.sockets:
                sock.listen()

        # Leaving the `with` block closed the port.
        ```

        Without a context manager, you must close the port yourself:

        ```python
        port = claim_port(8000)

        try:
            for sock in port.sockets:
                sock.listen()

        finally:
            port.close()
        ```
    """
    check_port_is_in_range(port)

    families = [AF_INET]

    # Without the IPv6-only option, an IPv6 socket can't be kept from claiming IPv4 too.
    if AF_INET6 is not None and IPV6_ONLY_OPTION is not None and has_ipv6():
        families.append(AF_INET6)

    # Closes every socket bound so far if any bind fails.
    with ExitStack() as stack:
        guards: list[socket] = []

        if _HOLD_WILDCARDS:  # pragma: macos-and-windows-only
            # Windows lets a socket that accepts IPv4 and IPv6 at once share its port
            # with sockets that accept only one. It can't be held alongside the other
            # wildcards, so it's bound and closed first.
            if AF_INET6 is not None and AF_INET6 in families and has_dualstack_ipv6():
                _bind(
                    AF_INET6,
                    _WILDCARDS[AF_INET6],
                    port,
                    dual_stack=True,
                ).close()

            guards = [
                stack.enter_context(
                    _bind(
                        family,
                        _WILDCARDS[family],
                        port,
                    ),
                )
                for family in families
            ]

        sockets = [
            stack.enter_context(
                _bind(
                    family,
                    _LOOPBACKS[family],
                    port,
                ),
            )
            for family in families
        ]

        # Every socket is bound, so stop the stack closing them. The claimed port owns
        # them now, and closes them all when it's closed.
        stack.pop_all()

    return ClaimedPort(
        port=port,
        sockets=sockets,
        guards=guards,
    )