Gecko [ v99win.today ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
rdtypes	[ DIR ]	drwxr-xr-x
__init__.py	1.56 KB	-rw-r--r--
_asyncbackend.py	1.53 KB	-rw-r--r--
_asyncio_backend.py	4.34 KB	-rw-r--r--
_curio_backend.py	3.33 KB	-rw-r--r--
_immutable_attr.py	2.92 KB	-rw-r--r--
_immutable_ctx.py	2.4 KB	-rw-r--r--
_trio_backend.py	3.68 KB	-rw-r--r--
asyncbackend.py	3 KB	-rw-r--r--
asyncquery.py	17.4 KB	-rw-r--r--
asyncresolver.py	8.71 KB	-rw-r--r--
dnssec.py	19.19 KB	-rw-r--r--
e164.py	3.6 KB	-rw-r--r--
edns.py	10.72 KB	-rw-r--r--
entropy.py	4.07 KB	-rw-r--r--
enum.py	2.75 KB	-rw-r--r--
exception.py	5.13 KB	-rw-r--r--
flags.py	2.58 KB	-rw-r--r--
grange.py	2.04 KB	-rw-r--r--
immutable.py	1.97 KB	-rw-r--r--
inet.py	4.74 KB	-rw-r--r--
ipv4.py	1.93 KB	-rw-r--r--
ipv6.py	5.92 KB	-rw-r--r--
message.py	52.73 KB	-rw-r--r--
name.py	31.54 KB	-rw-r--r--
namedict.py	3.83 KB	-rw-r--r--
node.py	6.42 KB	-rw-r--r--
opcode.py	2.58 KB	-rw-r--r--
py.typed	0 B	-rw-r--r--
query.py	40.46 KB	-rw-r--r--
rcode.py	3.9 KB	-rw-r--r--
rdata.py	24.32 KB	-rw-r--r--
rdataclass.py	2.81 KB	-rw-r--r--
rdataset.py	14.54 KB	-rw-r--r--
rdatatype.py	6.64 KB	-rw-r--r--
renderer.py	8.59 KB	-rw-r--r--
resolver.py	59.29 KB	-rw-r--r--
reversename.py	3.65 KB	-rw-r--r--
rrset.py	7.44 KB	-rw-r--r--
serial.py	3.53 KB	-rw-r--r--
set.py	7.64 KB	-rw-r--r--
tokenizer.py	21.86 KB	-rw-r--r--
transaction.py	16.93 KB	-rw-r--r--
tsig.py	10.93 KB	-rw-r--r--
tsigkeyring.py	2.43 KB	-rw-r--r--
ttl.py	2.58 KB	-rw-r--r--
update.py	10.55 KB	-rw-r--r--
version.py	1.55 KB	-rw-r--r--
versioned.py	15.91 KB	-rw-r--r--
wire.py	2.48 KB	-rw-r--r--
xfr.py	11.61 KB	-rw-r--r--
zone.py	34.25 KB	-rw-r--r--
zonefile.py	15.25 KB	-rw-r--r--

Code Editor : _asyncio_backend.py

# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license

"""asyncio library query support"""

import socket
import asyncio

import dns._asyncbackend
import dns.exception

def _get_running_loop():
    try:
        return asyncio.get_running_loop()
    except AttributeError:  # pragma: no cover
        return asyncio.get_event_loop()

class _DatagramProtocol:
    def __init__(self):
        self.transport = None
        self.recvfrom = None

def connection_made(self, transport):
        self.transport = transport

def datagram_received(self, data, addr):
        if self.recvfrom:
            self.recvfrom.set_result((data, addr))
            self.recvfrom = None

def error_received(self, exc):  # pragma: no cover
        if self.recvfrom and not self.recvfrom.done():
            self.recvfrom.set_exception(exc)

def connection_lost(self, exc):
        if self.recvfrom and not self.recvfrom.done():
            self.recvfrom.set_exception(exc)

def close(self):
        self.transport.close()

async def _maybe_wait_for(awaitable, timeout):
    if timeout:
        try:
            return await asyncio.wait_for(awaitable, timeout)
        except asyncio.TimeoutError:
            raise dns.exception.Timeout(timeout=timeout)
    else:
        return await awaitable

class DatagramSocket(dns._asyncbackend.DatagramSocket):
    def __init__(self, family, transport, protocol):
        self.family = family
        self.transport = transport
        self.protocol = protocol

async def sendto(self, what, destination, timeout):  # pragma: no cover
        # no timeout for asyncio sendto
        self.transport.sendto(what, destination)

async def recvfrom(self, size, timeout):
        # ignore size as there's no way I know to tell protocol about it
        done = _get_running_loop().create_future()
        assert self.protocol.recvfrom is None
        self.protocol.recvfrom = done
        await _maybe_wait_for(done, timeout)
        return done.result()

async def close(self):
        self.protocol.close()

async def getpeername(self):
        return self.transport.get_extra_info('peername')

async def getsockname(self):
        return self.transport.get_extra_info('sockname')

class StreamSocket(dns._asyncbackend.DatagramSocket):
    def __init__(self, af, reader, writer):
        self.family = af
        self.reader = reader
        self.writer = writer

async def sendall(self, what, timeout):
        self.writer.write(what)
        return await _maybe_wait_for(self.writer.drain(), timeout)

async def recv(self, count, timeout):
        return await _maybe_wait_for(self.reader.read(count),
                                     timeout)

async def close(self):
        self.writer.close()
        try:
            await self.writer.wait_closed()
        except AttributeError:  # pragma: no cover
            pass

async def getpeername(self):
        return self.writer.get_extra_info('peername')

async def getsockname(self):
        return self.writer.get_extra_info('sockname')

class Backend(dns._asyncbackend.Backend):
    def name(self):
        return 'asyncio'

async def make_socket(self, af, socktype, proto=0,
                          source=None, destination=None, timeout=None,
                          ssl_context=None, server_hostname=None):
        loop = _get_running_loop()
        if socktype == socket.SOCK_DGRAM:
            transport, protocol = await loop.create_datagram_endpoint(
                _DatagramProtocol, source, family=af,
                proto=proto)
            return DatagramSocket(af, transport, protocol)
        elif socktype == socket.SOCK_STREAM:
            (r, w) = await _maybe_wait_for(
                asyncio.open_connection(destination[0],
                                        destination[1],
                                        ssl=ssl_context,
                                        family=af,
                                        proto=proto,
                                        local_addr=source,
                                        server_hostname=server_hostname),
                timeout)
            return StreamSocket(af, r, w)
        raise NotImplementedError('unsupported socket ' +
                                  f'type {socktype}')  # pragma: no cover

async def sleep(self, interval):
        await asyncio.sleep(interval)

${this.title}

Code Editor : _asyncio_backend.py