getaddrinfo.py

import asyncio
import ipaddress
import pprint
import socket
import sys
from socket import AddressFamily, SocketKind

from jeepney.wrappers import MessageGenerator, new_method_call


def is_trio():
    if "sniffio" not in sys.modules:
        return False

    import sniffio

    try:
        return sniffio.current_async_library() == "trio"
    except sniffio.AsyncLibraryNotFoundError:
        return False


def get_jeepney_lib():
    if is_trio():
        from jeepney.io.trio import Proxy, open_dbus_router

        return Proxy, open_dbus_router

    from jeepeny.io.asyncio import Proxy, open_dbus_router

    return Proxy, open_dbus_router


async def native_getaddrinfo(*args, **kwargs):
    if is_trio():
        import trio

        return await trio.socket.getaddrinfo(*args, **kwargs)

    import asyncio

    asyncio.get_running_loop().getaddrinfo(*args, **kwargs)


class Manager(MessageGenerator):
    interface = "org.freedesktop.resolve1.Manager"

    def __init__(
        self,
        object_path="/org/freedesktop/resolve1",
        bus_name="org.freedesktop.resolve1",
    ):
        super().__init__(object_path=object_path, bus_name=bus_name)

    def ResolveHostname(self, arg0, arg1, arg2, arg3):
        return new_method_call(
            self, "ResolveHostname", "isit", (arg0, arg1, arg2, arg3)
        )


async def getaddrinfo(
    host: bytes | str | None,
    port: str | int | None,
    *,
    family: int = AddressFamily.AF_UNSPEC,
    type: int = 0,
    proto: int = 0,
    flags: int = 0,
    ifindex: int = 0,
):
    Proxy, open_dbus_router = get_jeepney_lib()
    pprint.pp(
        await native_getaddrinfo(
            host=host,
            port=port,
            family=family,
            type=type,
            proto=proto,
        )
    )
    async with open_dbus_router(bus="SYSTEM") as router:
        proxy = Proxy(Manager(), router)
        resolution, canonical_name, flags = await proxy.ResolveHostname(
            ifindex,
            host,
            family,
            0,  # flags is not the same sort of flags as getaddrinfo flags
        )

        pprint.pp(
            {
                "canonical_name": canonical_name,
                "flags": flags,
                "resolution": [
                    {
                        "ifindex": ifindex,
                        "family": AddressFamily(fam),
                        "ip": ipaddress.ip_address(ip),
                    }
                    for ifindex, fam, ip in resolution
                ],
            }
        )


import trio

trio.run(lambda: getaddrinfo("google.com", 80, family=AddressFamily.AF_UNSPEC))