# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license from typing import Dict import dns.exception # pylint: disable=unused-import from dns._asyncbackend import ( # noqa: F401 lgtm[py/unused-import] Backend, DatagramSocket, Socket, StreamSocket, ) # pylint: enable=unused-import _default_backend = None _backends: Dict[str, Backend] = {} # Allow sniffio import to be disabled for testing purposes _no_sniffio = False class AsyncLibraryNotFoundError(dns.exception.DNSException): pass def get_backend(name: str) -> Backend: """Get the specified asynchronous backend. *name*, a ``str``, the name of the backend. Currently the "trio" and "asyncio" backends are available. Raises NotImplementedError if an unknown backend name is specified. """ # pylint: disable=import-outside-toplevel,redefined-outer-name backend = _backends.get(name) if backend: return backend if name == "trio": import dns._trio_backend backend = dns._trio_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() -> str: """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. """ # pylint: disable=import-outside-toplevel 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") def get_default_backend() -> 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: str) -> Backend: """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