Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/resolver.py: 36%

1import asyncio 

2import socket 

3from typing import Any, Dict, List, Type, Union 

4 

5from .abc import AbstractResolver 

6 

7__all__ = ("ThreadedResolver", "AsyncResolver", "DefaultResolver") 

8 

9try: 

10 import aiodns 

11 

12 # aiodns_default = hasattr(aiodns.DNSResolver, 'gethostbyname') 

13except ImportError: # pragma: no cover 

14 aiodns = None 

15 

16aiodns_default = False 

17 

18 

19class ThreadedResolver(AbstractResolver): 

20 """Threaded resolver. 

21 

22 Uses an Executor for synchronous getaddrinfo() calls. 

23 concurrent.futures.ThreadPoolExecutor is used by default. 

24 """ 

25 

26 def __init__(self) -> None: 

27 self._loop = asyncio.get_running_loop() 

28 

29 async def resolve( 

30 self, hostname: str, port: int = 0, family: int = socket.AF_INET 

31 ) -> List[Dict[str, Any]]: 

32 infos = await self._loop.getaddrinfo( 

33 hostname, 

34 port, 

35 type=socket.SOCK_STREAM, 

36 family=family, 

37 flags=socket.AI_ADDRCONFIG, 

38 ) 

39 

40 hosts = [] 

41 for family, _, proto, _, address in infos: 

42 if family == socket.AF_INET6: 

43 if len(address) < 3: 

44 # IPv6 is not supported by Python build, 

45 # or IPv6 is not enabled in the host 

46 continue 

47 if address[3]: # type: ignore[misc] 

48 # This is essential for link-local IPv6 addresses. 

49 # LL IPv6 is a VERY rare case. Strictly speaking, we should use 

50 # getnameinfo() unconditionally, but performance makes sense. 

51 host, _port = socket.getnameinfo( 

52 address, socket.NI_NUMERICHOST | socket.NI_NUMERICSERV 

53 ) 

54 port = int(_port) 

55 else: 

56 host, port = address[:2] 

57 else: # IPv4 

58 assert family == socket.AF_INET 

59 host, port = address # type: ignore[misc] 

60 hosts.append( 

61 { 

62 "hostname": hostname, 

63 "host": host, 

64 "port": port, 

65 "family": family, 

66 "proto": proto, 

67 "flags": socket.AI_NUMERICHOST | socket.AI_NUMERICSERV, 

68 } 

69 ) 

70 

71 return hosts 

72 

73 async def close(self) -> None: 

74 pass 

75 

76 

77class AsyncResolver(AbstractResolver): 

78 """Use the `aiodns` package to make asynchronous DNS lookups""" 

79 

80 def __init__(self, *args: Any, **kwargs: Any) -> None: 

81 if aiodns is None: 

82 raise RuntimeError("Resolver requires aiodns library") 

83 

84 self._loop = asyncio.get_running_loop() 

85 self._resolver = aiodns.DNSResolver(*args, loop=self._loop, **kwargs) 

86 

87 async def resolve( 

88 self, host: str, port: int = 0, family: int = socket.AF_INET 

89 ) -> List[Dict[str, Any]]: 

90 try: 

91 resp = await self._resolver.gethostbyname(host, family) 

92 except aiodns.error.DNSError as exc: 

93 msg = exc.args[1] if len(exc.args) >= 1 else "DNS lookup failed" 

94 raise OSError(msg) from exc 

95 hosts = [] 

96 for address in resp.addresses: 

97 hosts.append( 

98 { 

99 "hostname": host, 

100 "host": address, 

101 "port": port, 

102 "family": family, 

103 "proto": 0, 

104 "flags": socket.AI_NUMERICHOST | socket.AI_NUMERICSERV, 

105 } 

106 ) 

107 

108 if not hosts: 

109 raise OSError("DNS lookup failed") 

110 

111 return hosts 

112 

113 async def close(self) -> None: 

114 self._resolver.cancel() 

115 

116 

117_DefaultType = Type[Union[AsyncResolver, ThreadedResolver]] 

118DefaultResolver: _DefaultType = AsyncResolver if aiodns_default else ThreadedResolver