Try Live
Add Docs
Rankings
Pricing
Enterprise
Docs
Install
Theme
Install
Docs
Pricing
Enterprise
More...
More...
Try Live
Rankings
Create API Key
Add Docs
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
Context
Skills
Chat
Benchmark
91.5
Suggestions
Latest
Show doc for...
Code
Info
Show Results
Context Summary (auto-generated)
Raw
Copy
Link
# 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.