Repository: yt-dlp/yt-dlp
File: yt_dlp/networking/_helper.py (Python)

Code before:


def _socket_connect(ip_addr, timeout, source_address):
    af, socktype, proto, canonname, sa = ip_addr
    sock = socket.socket(af, socktype, proto)
    try:
        if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:
            sock.settimeout(timeout)
        if source_address:
            sock.bind(source_address)
        sock.connect(sa)
        return sock
    except socket.error:
        sock.close()
        raise


def create_connection(address, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None, *, _create_socket_func=_socket_connect):
    # This is to workaround _create_connection() from socket where it will try all
    # address data from getaddrinfo() including IPv6. This filters the result from
    # getaddrinfo() based on the source_address value.
    # This is based on the cpython socket.create_connection() function.
    # https://github.com/python/cpython/blob/main/Lib/socket.py#L810
    host, port = address
    err = None
    ip_addrs = addrs = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)
    if source_address is not None:
        af = socket.AF_INET if '.' in source_address[0] else socket.AF_INET6
        ip_addrs = [addr for addr in addrs if addr[0] == af]
        if addrs and not ip_addrs:
            raise OSError(
                f"No remote IPv{4 if af == socket.AF_INET else 6} addresses available for connect. "
                f"Can't use '{source_address[0]}' as source address")

    for ip_addr in ip_addrs:
        try:
            sock = _create_socket_func(ip_addr, timeout, source_address)
            # Break explicitly a reference cycle
            err = None
            return sock
        except socket.error as e:
            err = e

    if err is not None:
        try:
            raise err
        finally:
            # Break explicitly a reference cycle
            err = None
    else:
        raise socket.error("getaddrinfo returns an empty list")

Review comment (suggestion):
```suggestion
    ip_addrs = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)
    if not ip_addrs:
        raise socket.error('getaddrinfo returns an empty list')
```
This seems easier to read than the fallthrough. This also enable us to use
```diff
-        ip_addrs = [addr for addr in addrs if addr[0] == af]
-        if addrs and not ip_addrs:
-            raise OSError(
-                f"No remote IPv{4 if af == socket.AF_INET else 6} addresses available for connect. "
-                f"Can't use '{source_address[0]}' as source address")
+        ip_addrs = [addr for addr in ip_addrs if addr[0] == af]
+        if not ip_addrs:
+            raise OSError(
+                f'No remote IPv{4 if af == socket.AF_INET else 6} addresses available for connect. '
+                f'Can\'t use "{source_address[0]}" as source address')
```

Code after:


def _socket_connect(ip_addr, timeout, source_address):
    af, socktype, proto, canonname, sa = ip_addr
    sock = socket.socket(af, socktype, proto)
    try:
        if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:
            sock.settimeout(timeout)
        if source_address:
            sock.bind(source_address)
        sock.connect(sa)
        return sock
    except socket.error:
        sock.close()
        raise


def create_connection(
    address,
    timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
    source_address=None,
    *,
    _create_socket_func=_socket_connect
):
    # Work around socket.create_connection() which tries all addresses from getaddrinfo() including IPv6.
    # This filters the addresses based on the given source_address.
    # Based on: https://github.com/python/cpython/blob/main/Lib/socket.py#L810
    host, port = address
    ip_addrs = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)
    if not ip_addrs:
        raise socket.error('getaddrinfo returns an empty list')
    if source_address is not None:
        af = socket.AF_INET if ':' not in source_address[0] else socket.AF_INET6
        ip_addrs = [addr for addr in ip_addrs if addr[0] == af]
        if not ip_addrs:
            raise OSError(
                f'No remote IPv{4 if af == socket.AF_INET else 6} addresses available for connect. '
                f'Can\'t use "{source_address[0]}" as source address')

    err = None
    for ip_addr in ip_addrs:
        try:
            sock = _create_socket_func(ip_addr, timeout, source_address)
            # Explicitly break __traceback__ reference cycle
            # https://bugs.python.org/issue36820
            err = None
            return sock
        except socket.error as e:
            err = e

    try: