Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/requests_toolbelt/adapters/source.py: 41%

1# -*- coding: utf-8 -*- 

2""" 

3requests_toolbelt.source_adapter 

4================================ 

5 

6This file contains an implementation of the SourceAddressAdapter originally 

7demonstrated on the Requests GitHub page. 

8""" 

9from requests.adapters import HTTPAdapter 

10 

11from .._compat import poolmanager, basestring 

12 

13 

14class SourceAddressAdapter(HTTPAdapter): 

15 """ 

16 A Source Address Adapter for Python Requests that enables you to choose the 

17 local address to bind to. This allows you to send your HTTP requests from a 

18 specific interface and IP address. 

19 

20 Two address formats are accepted. The first is a string: this will set the 

21 local IP address to the address given in the string, and will also choose a 

22 semi-random high port for the local port number. 

23 

24 The second is a two-tuple of the form (ip address, port): for example, 

25 ``('10.10.10.10', 8999)``. This will set the local IP address to the first 

26 element, and the local port to the second element. If ``0`` is used as the 

27 port number, a semi-random high port will be selected. 

28 

29 .. warning:: Setting an explicit local port can have negative interactions 

30 with connection-pooling in Requests: in particular, it risks 

31 the possibility of getting "Address in use" errors. The 

32 string-only argument is generally preferred to the tuple-form. 

33 

34 Example usage: 

35 

36 .. code-block:: python 

37 

38 import requests 

39 from requests_toolbelt.adapters.source import SourceAddressAdapter 

40 

41 s = requests.Session() 

42 s.mount('http://', SourceAddressAdapter('10.10.10.10')) 

43 s.mount('https://', SourceAddressAdapter(('10.10.10.10', 8999))) 

44 """ 

45 def __init__(self, source_address, **kwargs): 

46 if isinstance(source_address, basestring): 

47 self.source_address = (source_address, 0) 

48 elif isinstance(source_address, tuple): 

49 self.source_address = source_address 

50 else: 

51 raise TypeError( 

52 "source_address must be IP address string or (ip, port) tuple" 

53 ) 

54 

55 super(SourceAddressAdapter, self).__init__(**kwargs) 

56 

57 def init_poolmanager(self, connections, maxsize, block=False): 

58 self.poolmanager = poolmanager.PoolManager( 

59 num_pools=connections, 

60 maxsize=maxsize, 

61 block=block, 

62 source_address=self.source_address) 

63 

64 def proxy_manager_for(self, *args, **kwargs): 

65 kwargs['source_address'] = self.source_address 

66 return super(SourceAddressAdapter, self).proxy_manager_for( 

67 *args, **kwargs)