Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpcore/backends/auto.py: 44%

1import typing 

2from typing import Optional 

3 

4import sniffio 

5 

6from .base import SOCKET_OPTION, AsyncNetworkBackend, AsyncNetworkStream 

7 

8 

9class AutoBackend(AsyncNetworkBackend): 

10 async def _init_backend(self) -> None: 

11 if not (hasattr(self, "_backend")): 

12 backend = sniffio.current_async_library() 

13 if backend == "trio": 

14 from .trio import TrioBackend 

15 

16 self._backend: AsyncNetworkBackend = TrioBackend() 

17 else: 

18 from .asyncio import AsyncIOBackend 

19 

20 self._backend = AsyncIOBackend() 

21 

22 async def connect_tcp( 

23 self, 

24 host: str, 

25 port: int, 

26 timeout: Optional[float] = None, 

27 local_address: Optional[str] = None, 

28 socket_options: typing.Optional[typing.Iterable[SOCKET_OPTION]] = None, 

29 ) -> AsyncNetworkStream: 

30 await self._init_backend() 

31 return await self._backend.connect_tcp( 

32 host, 

33 port, 

34 timeout=timeout, 

35 local_address=local_address, 

36 socket_options=socket_options, 

37 ) 

38 

39 async def connect_unix_socket( 

40 self, 

41 path: str, 

42 timeout: Optional[float] = None, 

43 socket_options: typing.Optional[typing.Iterable[SOCKET_OPTION]] = None, 

44 ) -> AsyncNetworkStream: # pragma: nocover 

45 await self._init_backend() 

46 return await self._backend.connect_unix_socket( 

47 path, timeout=timeout, socket_options=socket_options 

48 ) 

49 

50 async def sleep(self, seconds: float) -> None: # pragma: nocover 

51 await self._init_backend() 

52 return await self._backend.sleep(seconds)