aiodns

import aiodns

from aiodns import DNSResolver

from aiodns import error
from aiodns.error import DNSError, ARES_ENOTFOUND, ARES_ETIMEOUT
import socket
import asyncio
from types import TracebackType
from typing import Optional, Sequence, Union, Iterable, Literal, Any
import pycares

import asyncio
import aiodns

async def main():
    # Create a DNS resolver
    resolver = aiodns.DNSResolver()
    
    try:
        # Perform A record query (returns list[pycares.ares_query_a_result])
        a_result = await resolver.query('google.com', 'A')
        print(f"A records: {[r.host for r in a_result]}")
        
        # Perform AAAA record query (returns list[pycares.ares_query_aaaa_result])
        aaaa_result = await resolver.query('google.com', 'AAAA')
        print(f"IPv6 addresses: {[r.host for r in aaaa_result]}")
        
        # CNAME query returns single result (not a list)
        cname_result = await resolver.query('www.github.com', 'CNAME')
        print(f"CNAME: {cname_result.cname}")
        
        # MX query returns list of results
        mx_result = await resolver.query('github.com', 'MX')
        print(f"MX records: {[(r.host, r.priority) for r in mx_result]}")
        
        # Hostname resolution with /etc/hosts support
        import socket
        host_result = await resolver.gethostbyname('localhost', socket.AF_INET)
        print(f"Localhost IP: {host_result.addresses}")
        
    except aiodns.error.DNSError as e:
        print(f"DNS error: {e}")
    finally:
        # Clean up resources
        await resolver.close()

# Using async context manager (automatic cleanup)
async def context_example():
    async with aiodns.DNSResolver() as resolver:
        result = await resolver.query('example.com', 'A')
        print(f"IP addresses: {[r.host for r in result]}")
        # resolver.close() called automatically

asyncio.run(main())

class DNSResolver:
    def __init__(
        self, 
        nameservers: Optional[Sequence[str]] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
        **kwargs: Any
    ) -> None:
        """
        Create a DNS resolver instance.
        
        Parameters:
        - nameservers: Optional list of DNS server IP addresses
        - loop: Optional asyncio event loop (defaults to current loop)
        - **kwargs: Additional options passed to pycares.Channel
        """
    
    # Type-specific query method overloads
    def query(self, host: str, qtype: Literal['A'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_a_result]]:
        """Query A records (IPv4 addresses)."""
    
    def query(self, host: str, qtype: Literal['AAAA'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_aaaa_result]]:
        """Query AAAA records (IPv6 addresses)."""
    
    def query(self, host: str, qtype: Literal['CAA'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_caa_result]]:
        """Query CAA records (Certificate Authority Authorization)."""
    
    def query(self, host: str, qtype: Literal['CNAME'], qclass: Optional[str] = None) -> asyncio.Future[pycares.ares_query_cname_result]:
        """Query CNAME record (Canonical Name) - returns single result."""
    
    def query(self, host: str, qtype: Literal['MX'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_mx_result]]:
        """Query MX records (Mail Exchange)."""
    
    def query(self, host: str, qtype: Literal['NAPTR'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_naptr_result]]:
        """Query NAPTR records (Name Authority Pointer)."""
    
    def query(self, host: str, qtype: Literal['NS'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_ns_result]]:
        """Query NS records (Name Server)."""
    
    def query(self, host: str, qtype: Literal['PTR'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_ptr_result]]:
        """Query PTR records (Pointer)."""
    
    def query(self, host: str, qtype: Literal['SOA'], qclass: Optional[str] = None) -> asyncio.Future[pycares.ares_query_soa_result]:
        """Query SOA record (Start of Authority) - returns single result."""
    
    def query(self, host: str, qtype: Literal['SRV'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_srv_result]]:
        """Query SRV records (Service)."""
    
    def query(self, host: str, qtype: Literal['TXT'], qclass: Optional[str] = None) -> asyncio.Future[list[pycares.ares_query_txt_result]]:
        """Query TXT records (Text)."""
    
    # Generic query method for runtime use
    def query(self, host: str, qtype: str, qclass: Optional[str] = None) -> Union[asyncio.Future[list[Any]], asyncio.Future[Any]]:
        """
        Perform DNS query for specified record type.
        
        Parameters:
        - host: Hostname or domain to query
        - qtype: Query type ('A', 'AAAA', 'CNAME', 'MX', 'TXT', 'PTR', 'SOA', 'SRV', 'NS', 'CAA', 'NAPTR', 'ANY')
        - qclass: Query class ('IN', 'CHAOS', 'HS', 'NONE', 'ANY'), defaults to 'IN'
        
        Returns:
        - asyncio.Future with query results (type varies by query type)
        - Most queries return list[ResultType], but CNAME and SOA return single ResultType
        
        Raises:
        - ValueError: Invalid query type or class
        - DNSError: DNS resolution errors
        """

def gethostbyname(
    self, 
    host: str, 
    family: socket.AddressFamily
) -> asyncio.Future[pycares.ares_host_result]:
    """
    Resolve hostname to IP address, checking /etc/hosts first.
    
    Parameters:
    - host: Hostname to resolve
    - family: Address family (socket.AF_INET or socket.AF_INET6)
    
    Returns:
    - asyncio.Future[pycares.ares_host_result] with host information
    """

def getaddrinfo(
    self,
    host: str,
    family: socket.AddressFamily = socket.AF_UNSPEC,
    port: Optional[int] = None,
    proto: int = 0,
    type: int = 0,
    flags: int = 0
) -> asyncio.Future[pycares.ares_addrinfo_result]:
    """
    Get address information for hostname (async equivalent of socket.getaddrinfo).
    
    Parameters:
    - host: Hostname to resolve
    - family: Address family (socket.AF_INET, socket.AF_INET6, or socket.AF_UNSPEC)
    - port: Port number (optional)
    - proto: Protocol (0 for any)
    - type: Socket type (0 for any)
    - flags: Additional flags
    
    Returns:
    - asyncio.Future[pycares.ares_addrinfo_result] with address information
    """

def gethostbyaddr(self, name: str) -> asyncio.Future[pycares.ares_host_result]:
    """
    Perform reverse DNS lookup from IP address to hostname.
    
    Parameters:
    - name: IP address string to resolve
    
    Returns:
    - asyncio.Future[pycares.ares_host_result] with hostname information
    """

def getnameinfo(
    self,
    sockaddr: Union[tuple[str, int], tuple[str, int, int, int]],
    flags: int = 0
) -> asyncio.Future[pycares.ares_nameinfo_result]:
    """
    Reverse name resolution from socket address.
    
    Parameters:
    - sockaddr: Socket address tuple (IPv4: (host, port), IPv6: (host, port, flow, scope))
    - flags: Additional flags
    
    Returns:
    - asyncio.Future[pycares.ares_nameinfo_result] with name information
    """

@property
def nameservers(self) -> Sequence[str]:
    """Get current DNS nameservers."""

@nameservers.setter  
def nameservers(self, value: Iterable[Union[str, bytes]]) -> None:
    """Set DNS nameservers."""

def cancel(self) -> None:
    """Cancel all pending DNS queries. All futures will receive DNSError with ARES_ECANCELLED."""

async def close(self) -> None:
    """
    Cleanly close the DNS resolver and release all resources.
    Must be called when resolver is no longer needed.
    """

def __del__(self) -> None:
    """
    Handle cleanup when the resolver is garbage collected.
    Automatically calls _cleanup() to release resources if resolver was not properly closed.
    Note: Explicit close() is preferred over relying on garbage collection.
    """

async def __aenter__(self) -> 'DNSResolver':
    """Enter async context manager."""

async def __aexit__(
    self,
    exc_type: Optional[type[BaseException]],
    exc_val: Optional[BaseException], 
    exc_tb: Optional[TracebackType]
) -> None:
    """Exit async context manager, automatically calls close()."""

class DNSError(Exception):
    """Base class for all DNS errors."""

# Error constants (integers)
ARES_SUCCESS: int
ARES_ENOTFOUND: int  # Domain name not found
ARES_EFORMERR: int   # Format error
ARES_ESERVFAIL: int  # Server failure
ARES_ENOTINITIALIZED: int  # Not initialized
ARES_EBADNAME: int   # Bad domain name
ARES_ETIMEOUT: int   # Timeout
ARES_ECONNREFUSED: int  # Connection refused
ARES_ENOMEM: int     # Out of memory
ARES_ECANCELLED: int # Query cancelled
ARES_EBADQUERY: int  # Bad query
ARES_EBADRESP: int   # Bad response
ARES_ENODATA: int    # No data
ARES_EBADFAMILY: int # Bad address family
ARES_EBADFLAGS: int  # Bad flags
ARES_EBADHINTS: int  # Bad hints
ARES_EBADSTR: int    # Bad string
ARES_EREFUSED: int   # Refused
ARES_ENOTIMP: int    # Not implemented
ARES_ESERVICE: int   # Service error
ARES_EFILE: int      # File error
ARES_EOF: int        # End of file
ARES_EDESTRUCTION: int  # Destruction
ARES_ELOADIPHLPAPI: int  # Load IP helper API error
ARES_EADDRGETNETWORKPARAMS: int  # Address get network params error

# A record results
pycares.ares_query_a_result:
    host: str  # IPv4 address

# AAAA record results  
pycares.ares_query_aaaa_result:
    host: str  # IPv6 address

# CNAME record result
pycares.ares_query_cname_result:
    cname: str  # Canonical name

# MX record results
pycares.ares_query_mx_result:
    host: str      # Mail server hostname
    priority: int  # Priority value

# TXT record results
pycares.ares_query_txt_result:
    text: bytes  # Text record data

# PTR record results
pycares.ares_query_ptr_result:
    name: str  # Pointer target name

# SOA record result
pycares.ares_query_soa_result:
    nsname: str    # Primary nameserver
    hostmaster: str # Responsible person email
    serial: int     # Serial number
    refresh: int    # Refresh interval
    retry: int      # Retry interval  
    expires: int    # Expiry time
    minttl: int     # Minimum TTL

# SRV record results
pycares.ares_query_srv_result:
    host: str      # Target hostname
    port: int      # Port number
    priority: int  # Priority
    weight: int    # Weight

# NS record results
pycares.ares_query_ns_result:
    host: str  # Nameserver hostname

# CAA record results
pycares.ares_query_caa_result:
    critical: int  # Critical flag
    property: bytes # Property name
    value: bytes    # Property value

# NAPTR record results
pycares.ares_query_naptr_result:
    order: int         # Order value
    preference: int    # Preference value
    flags: bytes       # Flags
    service: bytes     # Service
    regexp: bytes      # Regular expression
    replacement: bytes # Replacement

# Host resolution results
pycares.ares_host_result:
    name: str           # Primary hostname
    aliases: List[str]  # Hostname aliases
    addresses: List[str] # IP addresses

# Address info results
pycares.ares_addrinfo_result:
    nodes: List[ares_addrinfo_node]  # Address information nodes

# Address info node
pycares.ares_addrinfo_node:
    family: int       # Address family (socket.AF_INET, socket.AF_INET6)
    socktype: int     # Socket type
    protocol: int     # Protocol
    addr: tuple       # Address tuple (host, port) or (host, port, flow, scope)
    canonname: str    # Canonical name (optional)

# Name info results  
pycares.ares_nameinfo_result:
    node: str     # Hostname
    service: str  # Service name

# For non-thread-safe pycares builds on Windows:
import asyncio
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())