1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
import socket
import dns.inet
# This is a nullcontext for both sync and async
class NullContext:
def __init__(self, enter_result=None):
self.enter_result = enter_result
def __enter__(self):
return self.enter_result
def __exit__(self, exc_type, exc_value, traceback):
pass
async def __aenter__(self):
return self.enter_result
async def __aexit__(self, exc_type, exc_value, traceback):
pass
# This is handy, but should probably move somewhere else!
def low_level_address_tuple(af, high_level_address_tuple):
address, port = high_level_address_tuple
if af == dns.inet.AF_INET:
return (address, port)
elif af == dns.inet.AF_INET6:
ai_flags = socket.AI_NUMERICHOST
((*_, tup), *_) = socket.getaddrinfo(address, port, flags=ai_flags)
return tup
else:
raise NotImplementedError(f'unknown address family {af}')
# These are declared here so backends can import them without creating
# circular dependencies with dns.asyncbackend.
class Socket:
async def close(self):
pass
async def __aenter__(self):
pass
async def __aexit__(self, exc_type, exc_value, traceback):
await self.close()
class DatagramSocket(Socket):
async def sendto(self, what, destination, timeout):
pass
async def recvfrom(self, size, timeout):
pass
class StreamSocket(Socket):
async def sendall(self, what, destination, timeout):
pass
async def recv(self, size, timeout):
pass
class Backend:
def name(self):
return 'unknown'
async def make_socket(self, af, socktype, proto=0,
source=None, raw_source=None,
ssl_context=None, server_hostname=None):
raise NotImplementedError
|
