Demonstration of Python asyncio.getaddrinfo()

getaddrinfo_demo.py

#!/usr/bin/env python3

import asyncio
import socket
import sys

_AI_ADDRCONFIG = socket.AI_ADDRCONFIG
if hasattr(socket, "AI_MASK"):
	_AI_ADDRCONFIG &= socket.AI_MASK


async def resolve_hostname(hostname, port):
	loop = asyncio.get_event_loop()

	try:
		# Get address info using asyncio
		addrinfo_list = await loop.getaddrinfo(
			hostname,
			port,
			type=socket.SOCK_STREAM,
			family=socket.AF_UNSPEC,
			flags=_AI_ADDRCONFIG
		)
		return addrinfo_list
	except socket.gaierror as e:
		print(f"Error resolving {hostname}: {e}")
		return []
	except Exception as e:
		print(f"Unexpected error resolving {hostname}: {e}")
		return []


def format_address_info(addrinfo_list):
	if not addrinfo_list:
		return "No results found"

	result_lines = []
	for i, addrinfo in enumerate(addrinfo_list, 1):
		family, socktype, proto, canonname, sockaddr = addrinfo

		# Get family name
		family_name = {
			socket.AF_INET: 'IPv4',
			socket.AF_INET6: 'IPv6',
			socket.AF_UNSPEC: 'Unspecified'
		}.get(family, f'Unknown ({family})')

		# Get socket type name
		socktype_name = {
			socket.SOCK_STREAM: 'TCP',
			socket.SOCK_DGRAM: 'UDP',
			socket.SOCK_RAW: 'RAW'
		}.get(socktype, f'Unknown ({socktype})')

		# Format socket address
		if len(sockaddr) == 2:  # IPv4
			ip, port = sockaddr
			sockaddr_str = f"{ip}:{port}"
		elif len(sockaddr) == 4:  # IPv6
			ip, port, flowinfo, scope_id = sockaddr
			sockaddr_str = f"[{ip}]:{port}"
		else:
			sockaddr_str = str(sockaddr)

		result_lines.append(f"  {i}. Family: {family_name}")
		result_lines.append(f"     Type: {socktype_name}")
		result_lines.append(f"     Protocol: {proto}")
		result_lines.append(f"     Canonical Name: {canonname or 'N/A'}")
		result_lines.append(f"     Address: {sockaddr_str}")
		result_lines.append("")

	return "\n".join(result_lines)


async def main():
	addrinfo_list = await resolve_hostname('app.logman.io', 433)

	if addrinfo_list:
		print(f"Found {len(addrinfo_list)} result(s):")
		print(format_address_info(addrinfo_list))
	else:
		print("No results found.")

if __name__ == "__main__":
	try:
		asyncio.run(main())
	except KeyboardInterrupt:
		print("\nProgram interrupted by user.")
	except Exception as e:
		print(f"Error running program: {e}")
		sys.exit(1)