9bbeb35c08
Fixed an issue that prevented the password reset tokens from working. Added email templates for password reset success and new account creation. Added more dynamic email template support.
101 lines
2.7 KiB
Python
101 lines
2.7 KiB
Python
# 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
|