| from typing import Optional |
|
|
| import sniffio |
|
|
| from .base import AsyncNetworkBackend, AsyncNetworkStream |
|
|
|
|
| class AutoBackend(AsyncNetworkBackend): |
| async def _init_backend(self) -> None: |
| if not (hasattr(self, "_backend")): |
| backend = sniffio.current_async_library() |
| if backend == "trio": |
| from .trio import TrioBackend |
|
|
| self._backend: AsyncNetworkBackend = TrioBackend() |
| else: |
| from .asyncio import AsyncIOBackend |
|
|
| self._backend = AsyncIOBackend() |
|
|
| async def connect_tcp( |
| self, |
| host: str, |
| port: int, |
| timeout: Optional[float] = None, |
| local_address: Optional[str] = None, |
| ) -> AsyncNetworkStream: |
| await self._init_backend() |
| return await self._backend.connect_tcp( |
| host, port, timeout=timeout, local_address=local_address |
| ) |
|
|
| async def connect_unix_socket( |
| self, path: str, timeout: Optional[float] = None |
| ) -> AsyncNetworkStream: |
| await self._init_backend() |
| return await self._backend.connect_unix_socket(path, timeout=timeout) |
|
|
| async def sleep(self, seconds: float) -> None: |
| await self._init_backend() |
| return await self._backend.sleep(seconds) |
|
|
