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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
import dns.exception
from dns._asyncbackend import Socket, DatagramSocket, \
StreamSocket, Backend # noqa:
_default_backend = None
_backends = {}
# Allow sniffio import to be disabled for testing purposes
_no_sniffio = False
class AsyncLibraryNotFoundError(dns.exception.DNSException):
pass
def get_backend(name):
"""Get the specified asychronous backend.
*name*, a ``str``, the name of the backend. Currently the "trio",
"curio", and "asyncio" backends are available.
Raises NotImplementError if an unknown backend name is specified.
"""
backend = _backends.get(name)
if backend:
return backend
if name == 'trio':
import dns._trio_backend
backend = dns._trio_backend.Backend()
elif name == 'curio':
import dns._curio_backend
backend = dns._curio_backend.Backend()
elif name == 'asyncio':
import dns._asyncio_backend
backend = dns._asyncio_backend.Backend()
else:
raise NotImplementedError(f'unimplemented async backend {name}')
_backends[name] = backend
return backend
def sniff():
"""Attempt to determine the in-use asynchronous I/O library by using
the ``sniffio`` module if it is available.
Returns the name of the library, or raises AsyncLibraryNotFoundError
if the library cannot be determined.
"""
try:
if _no_sniffio:
raise ImportError
import sniffio
try:
return sniffio.current_async_library()
except sniffio.AsyncLibraryNotFoundError:
raise AsyncLibraryNotFoundError('sniffio cannot determine ' +
'async library')
except ImportError:
import asyncio
try:
asyncio.get_running_loop()
return 'asyncio'
except RuntimeError:
raise AsyncLibraryNotFoundError('no async library detected')
except AttributeError: # pragma: no cover
# we have to check current_task on 3.6
if not asyncio.Task.current_task():
raise AsyncLibraryNotFoundError('no async library detected')
return 'asyncio'
def get_default_backend():
"""Get the default backend, initializing it if necessary.
"""
if _default_backend:
return _default_backend
return set_default_backend(sniff())
def set_default_backend(name):
"""Set the default backend.
It's not normally necessary to call this method, as
``get_default_backend()`` will initialize the backend
appropriately in many cases. If ``sniffio`` is not installed, or
in testing situations, this function allows the backend to be set
explicitly.
"""
global _default_backend
_default_backend = get_backend(name)
return _default_backend
|
