golden hour

/opt/cloudlinux/venv/lib/python3.11/site-packages/aiohttp

File/Folder	Size	Actions
.hash	-
__init__.py	7.58 KB
__pycache__	-
_cparser.pxd	4.22 KB
_find_header.pxd	68 B
_headers.pxi	1.96 KB
_helpers.cpython-311-x86_64-linux-gnu.so	563.5 KB
_helpers.pyi	202 B
_helpers.pyx	1.02 KB
_http_parser.cpython-311-x86_64-linux-gnu.so	2.66 MB
_http_parser.pyx	27.4 KB
_http_writer.cpython-311-x86_64-linux-gnu.so	477.98 KB
_http_writer.pyx	4.47 KB
_websocket.cpython-311-x86_64-linux-gnu.so	249.84 KB
_websocket.pyx	1.52 KB
abc.py	5.37 KB
base_protocol.py	2.68 KB
client.py	46.17 KB
client_exceptions.py	9.19 KB
client_proto.py	8.45 KB
client_reqrep.py	38.75 KB
client_ws.py	10.75 KB
compression_utils.py	4.9 KB
connector.py	51.56 KB
cookiejar.py	13.69 KB
formdata.py	5.96 KB
hdrs.py	4.5 KB
helpers.py	29.55 KB
http.py	1.8 KB
http_exceptions.py	2.65 KB
http_parser.py	34.66 KB
http_websocket.py	26.09 KB
http_writer.py	5.79 KB
locks.py	1.11 KB
log.py	325 B
multipart.py	31.71 KB
payload.py	13.22 KB
payload_streamer.py	2.04 KB
py.typed	7 B
pytest_plugin.py	11.33 KB
resolver.py	4.95 KB
streams.py	20.35 KB
tcp_helpers.py	961 B
test_utils.py	19.71 KB
tracing.py	14.78 KB
typedefs.py	1.44 KB
web.py	18.81 KB
web_app.py	17.88 KB
web_exceptions.py	10.12 KB
web_fileresponse.py	11.15 KB
web_log.py	7.62 KB
web_middlewares.py	3.94 KB
web_protocol.py	22.5 KB
web_request.py	28.08 KB
web_response.py	27.08 KB
web_routedef.py	5.99 KB
web_runner.py	11.46 KB
web_server.py	2.53 KB
web_urldispatcher.py	39.12 KB
web_ws.py	18.21 KB
worker.py	7.78 KB

Edit: resolver.py

import asyncio
import socket
from typing import Any, Dict, List, Optional, Type, Union

from .abc import AbstractResolver
from .helpers import get_running_loop

__all__ = ("ThreadedResolver", "AsyncResolver", "DefaultResolver")

try:
    import aiodns

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

aiodns_default = False

class ThreadedResolver(AbstractResolver):
    """Threaded resolver.

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

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

async def resolve(
        self, hostname: str, port: int = 0, family: int = socket.AF_INET
    ) -> 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) < 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(
                {
                    "hostname": hostname,
                    "host": host,
                    "port": port,
                    "family": family,
                    "proto": proto,
                    "flags": socket.AI_NUMERICHOST | socket.AI_NUMERICSERV,
                }
            )

return hosts

async def close(self) -> None:
        pass

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

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

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

if not hasattr(self._resolver, "gethostbyname"):
            # 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
    ) -> 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) >= 1 else "DNS lookup failed"
            raise OSError(msg) from exc
        hosts = []
        for address in resp.addresses:
            hosts.append(
                {
                    "hostname": host,
                    "host": address,
                    "port": port,
                    "family": family,
                    "proto": 0,
                    "flags": socket.AI_NUMERICHOST | socket.AI_NUMERICSERV,
                }
            )

if not hosts:
            raise OSError("DNS lookup failed")

return hosts

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

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

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

if not hosts:
            raise OSError("DNS lookup failed")

return hosts

async def close(self) -> None:
        self._resolver.cancel()

_DefaultType = Type[Union[AsyncResolver, ThreadedResolver]]
DefaultResolver: _DefaultType = AsyncResolver if aiodns_default else ThreadedResolver