codekingpro/portable-devtools

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license2 3import dns._features4import dns.asyncbackend5 6if dns._features.have("doq"):7    import aioquic.quic.configuration  # type: ignore8 9    from dns._asyncbackend import NullContext10    from dns.quic._asyncio import (11        AsyncioQuicConnection,12        AsyncioQuicManager,13        AsyncioQuicStream,14    )15    from dns.quic._common import AsyncQuicConnection, AsyncQuicManager16    from dns.quic._sync import SyncQuicConnection, SyncQuicManager, SyncQuicStream17 18    have_quic = True19 20    def null_factory(21        *args,  # pylint: disable=unused-argument22        **kwargs,  # pylint: disable=unused-argument23    ):24        return NullContext(None)25 26    def _asyncio_manager_factory(27        context, *args, **kwargs  # pylint: disable=unused-argument28    ):29        return AsyncioQuicManager(*args, **kwargs)30 31    # We have a context factory and a manager factory as for trio we need to have32    # a nursery.33 34    _async_factories = {"asyncio": (null_factory, _asyncio_manager_factory)}35 36    if dns._features.have("trio"):37        import trio38 39        from dns.quic._trio import (  # pylint: disable=ungrouped-imports40            TrioQuicConnection,41            TrioQuicManager,42            TrioQuicStream,43        )44 45        def _trio_context_factory():46            return trio.open_nursery()47 48        def _trio_manager_factory(context, *args, **kwargs):49            return TrioQuicManager(context, *args, **kwargs)50 51        _async_factories["trio"] = (_trio_context_factory, _trio_manager_factory)52 53    def factories_for_backend(backend=None):54        if backend is None:55            backend = dns.asyncbackend.get_default_backend()56        return _async_factories[backend.name()]57 58else:  # pragma: no cover59    have_quic = False60 61    from typing import Any62 63    class AsyncQuicStream:  # type: ignore64        pass65 66    class AsyncQuicConnection:  # type: ignore67        async def make_stream(self) -> Any:68            raise NotImplementedError69 70    class SyncQuicStream:  # type: ignore71        pass72 73    class SyncQuicConnection:  # type: ignore74        def make_stream(self) -> Any:75            raise NotImplementedError76