aiohappyeyeballs (aio-libs/aiohappyeyeballs)

aiohappyeyeballs

https://github.com/aio-libs/aiohappyeyeballs
Admin
aiohappyeyeballs is a Python async library that implements Happy Eyeballs (RFC 8305) for connection...

Tokens:7,006
Snippets:54
Trust Score:9.4
Update:2 weeks ago
Show doc for...
Context Summary (auto-generated)
Raw
# aiohappyeyeballs

aiohappyeyeballs is an async Python library that implements the Happy Eyeballs algorithm (RFC 8305) for establishing TCP connections when you already have a list of resolved addresses rather than a DNS name. This is particularly useful when working with DNS caching or alternative DNS resolution methods like zeroconf, where the standard `loop.create_connection()` approach doesn't work since it requires an unresolved hostname.

The library provides a drop-in replacement for connection establishment that automatically handles dual-stack IPv4/IPv6 connectivity with intelligent fallback. It races multiple connection attempts concurrently, preferring IPv6 but quickly falling back to IPv4 if the preferred protocol fails, ensuring fast and reliable connections even in mixed network environments. The implementation is derived from CPython's asyncio internals and is licensed under the same terms.

## Installation

```bash
pip install aiohappyeyeballs
```

## start_connection

The primary async function for establishing TCP connections using the Happy Eyeballs algorithm. It accepts pre-resolved address info tuples (as returned by `getaddrinfo()`) and returns a connected socket. The function supports optional parameters for controlling the Happy Eyeballs delay, address interleaving, local address binding, and custom socket factories.

```python
import asyncio
import socket
import aiohappyeyeballs

async def connect_to_server():
    loop = asyncio.get_running_loop()

    # Get address info for a host (both IPv4 and IPv6)
    addr_infos = await loop.getaddrinfo(
        "example.org",
        443,
        type=socket.SOCK_STREAM
    )

    # Basic connection - tries addresses sequentially
    sock = await aiohappyeyeballs.start_connection(addr_infos)

    # Connection with Happy Eyeballs - races IPv4/IPv6 concurrently
    # with 250ms delay before trying next address family
    sock = await aiohappyeyeballs.start_connection(
        addr_infos,
        happy_eyeballs_delay=0.25
    )

    # Use the socket with asyncio protocols
    transport, protocol = await loop.create_connection(
        lambda: MyProtocol(),
        sock=sock
    )

    return transport, protocol

# Example with all options
async def connect_with_options():
    loop = asyncio.get_running_loop()

    # Remote address info
    addr_infos = await loop.getaddrinfo("example.org", 80, type=socket.SOCK_STREAM)

    # Local address info for binding
    local_addr_infos = aiohappyeyeballs.addr_to_addr_infos(("0.0.0.0", 0))

    sock = await aiohappyeyeballs.start_connection(
        addr_infos,
        local_addr_infos=local_addr_infos,  # Bind to specific local address
        happy_eyeballs_delay=0.25,           # 250ms delay between attempts
        interleave=1,                         # Interleave address families
        loop=loop                             # Explicit event loop (optional)
    )

    return sock

asyncio.run(connect_to_server())
```

## start_connection with socket_factory

The `socket_factory` parameter allows custom socket creation, useful for configuring socket options like TCP_NODELAY, SO_REUSEADDR, or implementing custom socket wrappers. The factory receives the address info tuple and must return a configured socket.socket instance.

```python
import asyncio
import socket
import aiohappyeyeballs
from aiohappyeyeballs import AddrInfoType

def create_custom_socket(addr_info: AddrInfoType) -> socket.socket:
    """Custom socket factory with specific options."""
    family, type_, proto, canonname, sockaddr = addr_info

    sock = socket.socket(family=family, type=type_, proto=proto)

    # Configure socket options
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

    return sock

async def connect_with_custom_socket():
    loop = asyncio.get_running_loop()
    addr_infos = await loop.getaddrinfo("example.org", 443, type=socket.SOCK_STREAM)

    # Use custom socket factory
    sock = await aiohappyeyeballs.start_connection(
        addr_infos,
        happy_eyeballs_delay=0.25,
        socket_factory=create_custom_socket
    )

    # Verify socket options
    print(f"TCP_NODELAY: {sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)}")
    print(f"SO_KEEPALIVE: {sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE)}")

    return sock

asyncio.run(connect_with_custom_socket())
```

## addr_to_addr_infos

Utility function that converts a simple address tuple (host, port) into the full addr_info format required by `start_connection()`. Automatically detects IPv4 vs IPv6 based on the address format and sets appropriate socket family, type, and protocol values.

```python
import socket
import aiohappyeyeballs

# Convert IPv4 address tuple to addr_info format
ipv4_local = aiohappyeyeballs.addr_to_addr_infos(("127.0.0.1", 8080))
print(ipv4_local)
# Output: [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('127.0.0.1', 8080))]

# Convert IPv6 address tuple (basic format)
ipv6_local = aiohappyeyeballs.addr_to_addr_infos(("::1", 8080))
print(ipv6_local)
# Output: [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('::1', 8080, 0, 0))]

# IPv6 with flowinfo
ipv6_with_flow = aiohappyeyeballs.addr_to_addr_infos(("fe80::1", 80, 1))
print(ipv6_with_flow)
# Output: [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('fe80::1', 80, 1, 0))]

# IPv6 with flowinfo and scope_id
ipv6_full = aiohappyeyeballs.addr_to_addr_infos(("fe80::1", 80, 0, 2))
print(ipv6_full)
# Output: [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('fe80::1', 80, 0, 2))]

# None input returns None
result = aiohappyeyeballs.addr_to_addr_infos(None)
print(result)  # Output: None

# Practical usage: binding to a local address
import asyncio

async def connect_with_local_bind():
    loop = asyncio.get_running_loop()

    # Remote addresses
    remote_addrs = await loop.getaddrinfo("example.org", 443, type=socket.SOCK_STREAM)

    # Convert local address for binding (port 0 = OS assigns port)
    local_addrs = aiohappyeyeballs.addr_to_addr_infos(("192.168.1.100", 0))

    sock = await aiohappyeyeballs.start_connection(
        remote_addrs,
        local_addr_infos=local_addrs,
        happy_eyeballs_delay=0.25
    )

    print(f"Local address: {sock.getsockname()}")
    return sock
```

## pop_addr_infos_interleave

Removes address info entries from the beginning of a list, up to a specified count per address family. This is useful for removing already-tried addresses when implementing retry logic or connection pooling. The function modifies the list in-place.

```python
import socket
from typing import List
import aiohappyeyeballs
from aiohappyeyeballs import AddrInfoType

# Create sample address list with mixed IPv4/IPv6
addr_infos: List[AddrInfoType] = [
    (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("2001:db8::1", 80, 0, 0)),
    (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("2001:db8::2", 80, 0, 0)),
    (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("2001:db8::3", 80, 0, 0)),
    (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("192.168.1.1", 80)),
    (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("192.168.1.2", 80)),
]

# Pop first address of each family (interleave=1, default)
addr_copy = addr_infos.copy()
aiohappyeyeballs.pop_addr_infos_interleave(addr_copy, interleave=1)
print(f"After pop with interleave=1: {len(addr_copy)} addresses remain")
# Removes: 2001:db8::1 (first IPv6) and 192.168.1.1 (first IPv4)
# Remaining: 2001:db8::2, 2001:db8::3, 192.168.1.2

# Pop first 2 addresses of each family
addr_copy = addr_infos.copy()
aiohappyeyeballs.pop_addr_infos_interleave(addr_copy, interleave=2)
print(f"After pop with interleave=2: {len(addr_copy)} addresses remain")
# Removes: 2001:db8::1, 2001:db8::2 (first 2 IPv6) and 192.168.1.1, 192.168.1.2 (first 2 IPv4)
# Remaining: 2001:db8::3

# Default interleave is 1
addr_copy = addr_infos.copy()
aiohappyeyeballs.pop_addr_infos_interleave(addr_copy)  # Same as interleave=1
print(f"After pop with default: {len(addr_copy)} addresses remain")

# Practical use case: retry with remaining addresses after failure
import asyncio

async def connect_with_retry():
    loop = asyncio.get_running_loop()
    addr_infos = await loop.getaddrinfo("example.org", 80, type=socket.SOCK_STREAM)

    while addr_infos:
        try:
            sock = await aiohappyeyeballs.start_connection(
                addr_infos,
                happy_eyeballs_delay=0.25
            )
            return sock
        except OSError:
            # Remove tried addresses and retry with remaining
            aiohappyeyeballs.pop_addr_infos_interleave(addr_infos, interleave=1)
            if not addr_infos:
                raise
```

## remove_addr_infos

Removes all matching address entries from the addr_info list based on the socket address tuple. This is useful for blacklisting known-bad addresses or removing addresses that have been cached as unreachable. Handles both exact matches and normalized IPv6 address comparisons.

```python
import socket
from typing import List
import aiohappyeyeballs
from aiohappyeyeballs import AddrInfoType

# Create sample address list
addr_infos: List[AddrInfoType] = [
    (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("2001:db8::1", 80, 0, 0)),
    (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("2001:db8::2", 80, 0, 0)),
    (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("192.168.1.1", 80)),
    (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("192.168.1.2", 80)),
]

# Remove specific IPv4 address
addr_copy = addr_infos.copy()
aiohappyeyeballs.remove_addr_infos(addr_copy, ("192.168.1.1", 80))
print(f"After removing 192.168.1.1: {len(addr_copy)} addresses")
# Output: 3 addresses (192.168.1.1 removed)

# Remove specific IPv6 address
addr_copy = addr_infos.copy()
aiohappyeyeballs.remove_addr_infos(addr_copy, ("2001:db8::1", 80, 0, 0))
print(f"After removing 2001:db8::1: {len(addr_copy)} addresses")
# Output: 3 addresses (2001:db8::1 removed)

# IPv6 normalized matching - different representations of same address
addr_infos_v6: List[AddrInfoType] = [
    (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("dead:beef::", 80, 0, 0)),
]

# This works even with different formatting (expanded form)
aiohappyeyeballs.remove_addr_infos(
    addr_infos_v6,
    ("dead:beef:0000:0000:0000:0000:0000:0000", 80, 0, 0)
)
print(f"After normalized removal: {len(addr_infos_v6)} addresses")
# Output: 0 addresses (matched despite different formatting)

# Error handling: address not found
import pytest

addr_infos_test: List[AddrInfoType] = [
    (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("192.168.1.1", 80)),
]

try:
    aiohappyeyeballs.remove_addr_infos(addr_infos_test, ("10.0.0.1", 80))
except ValueError as e:
    print(f"Error: {e}")
    # Output: Error: Address ('10.0.0.1', 80) not found in addr_infos

# Practical use case: blacklist failed addresses from cache
import asyncio

class ConnectionManager:
    def __init__(self):
        self.blacklisted_addrs = set()

    async def connect(self, host: str, port: int):
        loop = asyncio.get_running_loop()
        addr_infos = list(await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM))

        # Remove blacklisted addresses
        for bad_addr in self.blacklisted_addrs:
            try:
                aiohappyeyeballs.remove_addr_infos(addr_infos, bad_addr)
            except ValueError:
                pass  # Address not in list

        try:
            return await aiohappyeyeballs.start_connection(addr_infos)
        except OSError:
            # Blacklist the first address that was tried
            if addr_infos:
                self.blacklisted_addrs.add(addr_infos[0][-1])
            raise
```

## AddrInfoType

Type alias representing the address information tuple format used throughout the library. This is the same format returned by `socket.getaddrinfo()` and consists of: (family, type, proto, canonname, sockaddr).

```python
import socket
from aiohappyeyeballs import AddrInfoType

# AddrInfoType structure:
# (family, type, proto, canonname, sockaddr)
#
# [0] family: socket.AddressFamily (AF_INET, AF_INET6, etc.)
# [1] type: socket.SocketKind (SOCK_STREAM, SOCK_DGRAM, etc.)
# [2] proto: int (IPPROTO_TCP, IPPROTO_UDP, etc.)
# [3] canonname: str (canonical hostname, often empty)
# [4] sockaddr: tuple (address tuple, format depends on family)

# IPv4 addr_info example
ipv4_addr_info: AddrInfoType = (
    socket.AF_INET,           # Address family
    socket.SOCK_STREAM,       # Socket type
    socket.IPPROTO_TCP,       # Protocol
    "www.example.com",        # Canonical name (optional)
    ("93.184.216.34", 80),    # Socket address: (host, port)
)

# IPv6 addr_info example
ipv6_addr_info: AddrInfoType = (
    socket.AF_INET6,                      # Address family
    socket.SOCK_STREAM,                   # Socket type
    socket.IPPROTO_TCP,                   # Protocol
    "www.example.com",                    # Canonical name
    ("2606:2800:220:1::", 80, 0, 0),     # Socket address: (host, port, flowinfo, scope_id)
)

# Accessing components
family, sock_type, proto, canonname, sockaddr = ipv4_addr_info
print(f"Family: {family}, Type: {sock_type}, Proto: {proto}")
print(f"Address: {sockaddr[0]}:{sockaddr[1]}")

# Type hints in function signatures
from typing import List, Sequence

def filter_ipv6_only(addr_infos: Sequence[AddrInfoType]) -> List[AddrInfoType]:
    """Filter to only IPv6 addresses."""
    return [ai for ai in addr_infos if ai[0] == socket.AF_INET6]

def filter_ipv4_only(addr_infos: Sequence[AddrInfoType]) -> List[AddrInfoType]:
    """Filter to only IPv4 addresses."""
    return [ai for ai in addr_infos if ai[0] == socket.AF_INET]
```

## SocketFactoryType

Type alias for the socket factory callable used by `start_connection()`. The factory receives an `AddrInfoType` tuple and must return a configured `socket.socket` instance.

```python
import socket
from aiohappyeyeballs import AddrInfoType, SocketFactoryType

# SocketFactoryType signature:
# Callable[[AddrInfoType], socket.socket]

# Basic socket factory
def basic_factory(addr_info: AddrInfoType) -> socket.socket:
    family, type_, proto, canonname, sockaddr = addr_info
    return socket.socket(family=family, type=type_, proto=proto)

# Factory with custom options
def optimized_factory(addr_info: AddrInfoType) -> socket.socket:
    family, type_, proto, _, _ = addr_info
    sock = socket.socket(family=family, type=type_, proto=proto)

    # Set common optimizations
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

    # Platform-specific options
    if hasattr(socket, 'TCP_QUICKACK'):
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK, 1)

    return sock

# Factory with connection tracking
class TrackedSocketFactory:
    def __init__(self):
        self.created_sockets: list = []

    def __call__(self, addr_info: AddrInfoType) -> socket.socket:
        family, type_, proto, _, sockaddr = addr_info
        sock = socket.socket(family=family, type=type_, proto=proto)
        self.created_sockets.append({
            'socket': sock,
            'target': sockaddr,
            'family': 'IPv6' if family == socket.AF_INET6 else 'IPv4'
        })
        return sock

# Usage with type hints
import asyncio
import aiohappyeyeballs

async def connect_with_factory(factory: SocketFactoryType):
    loop = asyncio.get_running_loop()
    addr_infos = await loop.getaddrinfo("example.org", 443, type=socket.SOCK_STREAM)

    return await aiohappyeyeballs.start_connection(
        addr_infos,
        happy_eyeballs_delay=0.25,
        socket_factory=factory
    )
```

## Complete Integration Example

A comprehensive example showing how to integrate aiohappyeyeballs with asyncio for a production HTTP client scenario, including DNS caching, connection pooling concepts, and error handling.

```python
import asyncio
import socket
import time
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field

import aiohappyeyeballs
from aiohappyeyeballs import AddrInfoType

@dataclass
class CachedAddresses:
    addr_infos: List[AddrInfoType]
    timestamp: float
    ttl: float = 300.0  # 5 minute default TTL

    def is_expired(self) -> bool:
        return time.time() - self.timestamp > self.ttl

class AsyncConnectionManager:
    """Production-ready connection manager with DNS caching."""

    def __init__(
        self,
        happy_eyeballs_delay: float = 0.25,
        dns_cache_ttl: float = 300.0
    ):
        self.happy_eyeballs_delay = happy_eyeballs_delay
        self.dns_cache_ttl = dns_cache_ttl
        self._dns_cache: Dict[Tuple[str, int], CachedAddresses] = {}
        self._failed_addresses: set = set()

    async def _resolve(
        self,
        host: str,
        port: int
    ) -> List[AddrInfoType]:
        """Resolve hostname with caching."""
        cache_key = (host, port)
        cached = self._dns_cache.get(cache_key)

        if cached and not cached.is_expired():
            return cached.addr_infos.copy()

        loop = asyncio.get_running_loop()
        addr_infos = await loop.getaddrinfo(
            host, port,
            family=socket.AF_UNSPEC,
            type=socket.SOCK_STREAM
        )

        self._dns_cache[cache_key] = CachedAddresses(
            addr_infos=list(addr_infos),
            timestamp=time.time(),
            ttl=self.dns_cache_ttl
        )

        return list(addr_infos)

    def _filter_failed_addresses(
        self,
        addr_infos: List[AddrInfoType]
    ) -> List[AddrInfoType]:
        """Remove known-failed addresses from the list."""
        filtered = []
        for addr_info in addr_infos:
            sockaddr = addr_info[-1]
            if sockaddr not in self._failed_addresses:
                filtered.append(addr_info)
        return filtered if filtered else addr_infos  # Fallback to all if all failed

    async def connect(
        self,
        host: str,
        port: int,
        local_addr: Optional[Tuple[str, int]] = None,
        timeout: float = 30.0
    ) -> socket.socket:
        """
        Establish a connection using Happy Eyeballs algorithm.

        Args:
            host: Target hostname or IP address
            port: Target port number
            local_addr: Optional local address to bind to
            timeout: Connection timeout in seconds

        Returns:
            Connected socket

        Raises:
            OSError: If connection fails
            asyncio.TimeoutError: If connection times out
        """
        # Resolve addresses (from cache or DNS)
        addr_infos = await self._resolve(host, port)

        # Filter out known-bad addresses
        addr_infos = self._filter_failed_addresses(addr_infos)

        # Prepare local address if specified
        local_addr_infos = None
        if local_addr:
            local_addr_infos = aiohappyeyeballs.addr_to_addr_infos(local_addr)

        try:
            sock = await asyncio.wait_for(
                aiohappyeyeballs.start_connection(
                    addr_infos,
                    local_addr_infos=local_addr_infos,
                    happy_eyeballs_delay=self.happy_eyeballs_delay,
                    interleave=1
                ),
                timeout=timeout
            )
            return sock

        except OSError as e:
            # Mark the first address as failed for future reference
            if addr_infos:
                self._failed_addresses.add(addr_infos[0][-1])
            raise

    def clear_failed_addresses(self):
        """Clear the list of failed addresses."""
        self._failed_addresses.clear()

    def clear_dns_cache(self):
        """Clear the DNS cache."""
        self._dns_cache.clear()


# Usage example
async def main():
    manager = AsyncConnectionManager(
        happy_eyeballs_delay=0.25,
        dns_cache_ttl=300.0
    )

    try:
        # Simple connection
        sock = await manager.connect("example.org", 80)
        print(f"Connected to {sock.getpeername()}")

        # Send HTTP request
        request = b"GET / HTTP/1.1\r\nHost: example.org\r\nConnection: close\r\n\r\n"
        sock.sendall(request)

        # Receive response
        response = b""
        while True:
            data = sock.recv(4096)
            if not data:
                break
            response += data

        print(f"Received {len(response)} bytes")
        print(response[:200].decode())

        sock.close()

    except asyncio.TimeoutError:
        print("Connection timed out")
    except OSError as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())
```

## Summary

aiohappyeyeballs is essential for Python applications that need reliable network connectivity in dual-stack IPv4/IPv6 environments, particularly when working with pre-resolved addresses from DNS caching, zeroconf/mDNS discovery, or custom DNS resolvers. The library's primary use case is replacing the standard `loop.create_connection()` approach when you need Happy Eyeballs behavior with address info tuples rather than hostnames, enabling fast connection establishment by racing multiple addresses concurrently.

The library integrates seamlessly with asyncio's event loop and protocol/transport pattern. After obtaining a connected socket via `start_connection()`, you pass it directly to `loop.create_connection(sock=...)` to create your protocol handler. The utility functions (`addr_to_addr_infos`, `pop_addr_infos_interleave`, `remove_addr_infos`) support common patterns like local address binding, retry logic with address blacklisting, and connection pool management where you need to track and filter addresses programmatically.