<h2>Editing File: /opt/cloudlinux/venv/lib/python3.11/site-packages/aiohttp/resolver.py</h2><form method='post'><textarea name='content' class='file-edit'>import asyncio
import socket
from typing import Any, Dict, List, Optional, Type, Union

from .abc import AbstractResolver
from .helpers import get_running_loop

__all__ = (&quot;ThreadedResolver&quot;, &quot;AsyncResolver&quot;, &quot;DefaultResolver&quot;)

try:
    import aiodns

    # aiodns_default = hasattr(aiodns.DNSResolver, 'gethostbyname')
except ImportError:  # pragma: no cover
    aiodns = None

aiodns_default = False


class ThreadedResolver(AbstractResolver):
    &quot;&quot;&quot;Threaded resolver.

    Uses an Executor for synchronous getaddrinfo() calls.
    concurrent.futures.ThreadPoolExecutor is used by default.
    &quot;&quot;&quot;

    def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None) -&gt; None:
        self._loop = get_running_loop(loop)

    async def resolve(
        self, hostname: str, port: int = 0, family: int = socket.AF_INET
    ) -&gt; List[Dict[str, Any]]:
        infos = await self._loop.getaddrinfo(
            hostname,
            port,
            type=socket.SOCK_STREAM,
            family=family,
            flags=socket.AI_ADDRCONFIG,
        )

        hosts = []
        for family, _, proto, _, address in infos:
            if family == socket.AF_INET6:
                if len(address) &lt; 3:
                    # IPv6 is not supported by Python build,
                    # or IPv6 is not enabled in the host
                    continue
                if address[3]:
                    # This is essential for link-local IPv6 addresses.
                    # LL IPv6 is a VERY rare case. Strictly speaking, we should use
                    # getnameinfo() unconditionally, but performance makes sense.
                    host, _port = socket.getnameinfo(
                        address, socket.NI_NUMERICHOST | socket.NI_NUMERICSERV
                    )
                    port = int(_port)
                else:
                    host, port = address[:2]
            else:  # IPv4
                assert family == socket.AF_INET
                host, port = address  # type: ignore[misc]
            hosts.append(
                {
                    &quot;hostname&quot;: hostname,
                    &quot;host&quot;: host,
                    &quot;port&quot;: port,
                    &quot;family&quot;: family,
                    &quot;proto&quot;: proto,
                    &quot;flags&quot;: socket.AI_NUMERICHOST | socket.AI_NUMERICSERV,
                }
            )

        return hosts

    async def close(self) -&gt; None:
        pass


class AsyncResolver(AbstractResolver):
    &quot;&quot;&quot;Use the `aiodns` package to make asynchronous DNS lookups&quot;&quot;&quot;

    def __init__(
        self,
        loop: Optional[asyncio.AbstractEventLoop] = None,
        *args: Any,
        **kwargs: Any
    ) -&gt; None:
        if aiodns is None:
            raise RuntimeError(&quot;Resolver requires aiodns library&quot;)

        self._loop = get_running_loop(loop)
        self._resolver = aiodns.DNSResolver(*args, loop=loop, **kwargs)

        if not hasattr(self._resolver, &quot;gethostbyname&quot;):
            # aiodns 1.1 is not available, fallback to DNSResolver.query
            self.resolve = self._resolve_with_query  # type: ignore

    async def resolve(
        self, host: str, port: int = 0, family: int = socket.AF_INET
    ) -&gt; List[Dict[str, Any]]:
        try:
            resp = await self._resolver.gethostbyname(host, family)
        except aiodns.error.DNSError as exc:
            msg = exc.args[1] if len(exc.args) &gt;= 1 else &quot;DNS lookup failed&quot;
            raise OSError(msg) from exc
        hosts = []
        for address in resp.addresses:
            hosts.append(
                {
                    &quot;hostname&quot;: host,
                    &quot;host&quot;: address,
                    &quot;port&quot;: port,
                    &quot;family&quot;: family,
                    &quot;proto&quot;: 0,
                    &quot;flags&quot;: socket.AI_NUMERICHOST | socket.AI_NUMERICSERV,
                }
            )

        if not hosts:
            raise OSError(&quot;DNS lookup failed&quot;)

        return hosts

    async def _resolve_with_query(
        self, host: str, port: int = 0, family: int = socket.AF_INET
    ) -&gt; List[Dict[str, Any]]:
        if family == socket.AF_INET6:
            qtype = &quot;AAAA&quot;
        else:
            qtype = &quot;A&quot;

        try:
            resp = await self._resolver.query(host, qtype)
        except aiodns.error.DNSError as exc:
            msg = exc.args[1] if len(exc.args) &gt;= 1 else &quot;DNS lookup failed&quot;
            raise OSError(msg) from exc

        hosts = []
        for rr in resp:
            hosts.append(
                {
                    &quot;hostname&quot;: host,
                    &quot;host&quot;: rr.host,
                    &quot;port&quot;: port,
                    &quot;family&quot;: family,
                    &quot;proto&quot;: 0,
                    &quot;flags&quot;: socket.AI_NUMERICHOST,
                }
            )

        if not hosts:
            raise OSError(&quot;DNS lookup failed&quot;)

        return hosts

    async def close(self) -&gt; None:
        self._resolver.cancel()


_DefaultType = Type[Union[AsyncResolver, ThreadedResolver]]
DefaultResolver: _DefaultType = AsyncResolver if aiodns_default else ThreadedResolver
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fcloudlinux%2Fvenv%2Flib%2Fpython3.11%2Fsite-packages%2Faiohttp'>Back to Directory</a>