Pycares: Asynchronous DNS Resolver

5.0.1 · active · verified Sun Apr 05

Pycares is a Python module that provides an asynchronous interface to c-ares, a C library for performing DNS requests and name resolutions. It enables non-blocking DNS lookups, making it suitable for high-performance network applications. The library is actively maintained, currently at version 5.0.1, with regular releases addressing bug fixes and introducing new features.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to perform asynchronous DNS queries for A and MX records using `pycares`. It sets up a `Channel` and makes two queries with a shared callback function. A basic `select`-based loop is used to process file descriptors and handle the asynchronous responses. In a production environment, `pycares` is typically integrated with a more robust event loop like `asyncio` (via `aiodns`), `Tornado`, or `Gevent`.

import pycares
import socket

def callback(result, error):
    if error:
        print(f"Error: {error}")
        return
    if result:
        for record in result.answer:
            if record.type == pycares.QUERY_TYPE_A:
                print(f"A record for {record.name}: {record.data.addr}")
            elif record.type == pycares.QUERY_TYPE_AAAA:
                print(f"AAAA record for {record.name}: {record.data.addr}")
            elif record.type == pycares.QUERY_TYPE_MX:
                print(f"MX record for {record.name}: priority={record.data.priority}, exchange={record.data.exchange}")
            # Add other record types as needed
    else:
        print("No records found.")


# Using a simple select-based event loop
channel = pycares.Channel(timeout=5.0)

# Query for A records
channel.query("google.com", pycares.QUERY_TYPE_A, callback=callback)

# Query for MX records
channel.query("example.com", pycares.QUERY_TYPE_MX, callback=callback)

# Basic event loop processing
while True:
    read_fds, write_fds = channel.getsockname()
    if not read_fds and not write_fds:
        break
    
    # In a real application, use an actual event loop (e.g., asyncio, Tornado, Gevent)
    # For this simple example, we block briefly
    try:
        rlist, wlist, xlist = socket.select(read_fds, write_fds, [], 1.0)
    except socket.error as e:
        print(f"Socket error in select: {e}")
        break

    channel.process_fd(rlist, wlist)

view raw JSON →