codekingpro/portable-devtools

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license2 3from typing import Dict4 5import dns.exception6 7# pylint: disable=unused-import8from dns._asyncbackend import (  # noqa: F401  lgtm[py/unused-import]9    Backend,10    DatagramSocket,11    Socket,12    StreamSocket,13)14 15# pylint: enable=unused-import16 17_default_backend = None18 19_backends: Dict[str, Backend] = {}20 21# Allow sniffio import to be disabled for testing purposes22_no_sniffio = False23 24 25class AsyncLibraryNotFoundError(dns.exception.DNSException):26    pass27 28 29def get_backend(name: str) -> Backend:30    """Get the specified asynchronous backend.31 32    *name*, a ``str``, the name of the backend.  Currently the "trio"33    and "asyncio" backends are available.34 35    Raises NotImplementedError if an unknown backend name is specified.36    """37    # pylint: disable=import-outside-toplevel,redefined-outer-name38    backend = _backends.get(name)39    if backend:40        return backend41    if name == "trio":42        import dns._trio_backend43 44        backend = dns._trio_backend.Backend()45    elif name == "asyncio":46        import dns._asyncio_backend47 48        backend = dns._asyncio_backend.Backend()49    else:50        raise NotImplementedError(f"unimplemented async backend {name}")51    _backends[name] = backend52    return backend53 54 55def sniff() -> str:56    """Attempt to determine the in-use asynchronous I/O library by using57    the ``sniffio`` module if it is available.58 59    Returns the name of the library, or raises AsyncLibraryNotFoundError60    if the library cannot be determined.61    """62    # pylint: disable=import-outside-toplevel63    try:64        if _no_sniffio:65            raise ImportError66        import sniffio67 68        try:69            return sniffio.current_async_library()70        except sniffio.AsyncLibraryNotFoundError:71            raise AsyncLibraryNotFoundError("sniffio cannot determine async library")72    except ImportError:73        import asyncio74 75        try:76            asyncio.get_running_loop()77            return "asyncio"78        except RuntimeError:79            raise AsyncLibraryNotFoundError("no async library detected")80 81 82def get_default_backend() -> Backend:83    """Get the default backend, initializing it if necessary."""84    if _default_backend:85        return _default_backend86 87    return set_default_backend(sniff())88 89 90def set_default_backend(name: str) -> Backend:91    """Set the default backend.92 93    It's not normally necessary to call this method, as94    ``get_default_backend()`` will initialize the backend95    appropriately in many cases.  If ``sniffio`` is not installed, or96    in testing situations, this function allows the backend to be set97    explicitly.98    """99    global _default_backend100    _default_backend = get_backend(name)101    return _default_backend102