Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/looker_sdk/rtl/requests_transport.py: 50%

1# The MIT License (MIT) 

2# 

3# Copyright (c) 2019 Looker Data Sciences, Inc. 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy 

6# of this software and associated documentation files (the "Software"), to deal 

7# in the Software without restriction, including without limitation the rights 

8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

9# copies of the Software, and to permit persons to whom the Software is 

10# furnished to do so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in 

13# all copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 

21# THE SOFTWARE. 

22 

23"""Transport implementation using requests package.""" 

24 

25import logging 

26from typing import cast, Callable, Dict, MutableMapping, Optional 

27 

28import requests 

29 

30from looker_sdk.rtl import transport 

31 

32 

33class RequestsTransport(transport.Transport): 

34 """RequestsTransport implementation of Transport.""" 

35 

36 def __init__( 

37 self, settings: transport.PTransportSettings, session: requests.Session 

38 ): 

39 self.settings = settings 

40 headers: Dict[str, str] = {transport.LOOKER_API_ID: settings.agent_tag} 

41 if settings.headers: 

42 headers.update(settings.headers) 

43 session.headers.update(headers) 

44 session.verify = settings.verify_ssl 

45 self.session = session 

46 self.logger = logging.getLogger(__name__) 

47 

48 @classmethod 

49 def configure(cls, settings: transport.PTransportSettings) -> transport.Transport: 

50 return cls(settings, requests.Session()) 

51 

52 def request( 

53 self, 

54 method: transport.HttpMethod, 

55 path: str, 

56 query_params: Optional[MutableMapping[str, str]] = None, 

57 body: Optional[bytes] = None, 

58 authenticator: transport.TAuthenticator = None, 

59 transport_options: Optional[transport.TransportOptions] = None, 

60 ) -> transport.Response: 

61 

62 headers = {} 

63 timeout = self.settings.timeout 

64 if authenticator: 

65 headers.update(authenticator(transport_options or {})) 

66 if transport_options: 

67 if transport_options.get("headers"): 

68 headers.update(transport_options["headers"]) 

69 if transport_options.get("timeout"): 

70 timeout = transport_options["timeout"] 

71 self.logger.info("%s(%s)", method.name, path) 

72 try: 

73 resp = self.session.request( 

74 method.name, 

75 path, 

76 auth=NullAuth(), 

77 params=query_params, 

78 data=body, 

79 headers=headers, 

80 timeout=timeout, 

81 ) 

82 except IOError as exc: 

83 ret = transport.Response( 

84 False, 

85 bytes(str(exc), encoding="utf-8"), 

86 transport.ResponseMode.STRING, 

87 ) 

88 else: 

89 ret = transport.Response( 

90 resp.ok, 

91 resp.content, 

92 transport.response_mode(resp.headers.get("content-type")), 

93 ) 

94 encoding = cast( 

95 Optional[str], requests.utils.get_encoding_from_headers(resp.headers) 

96 ) 

97 if encoding: 

98 ret.encoding = encoding 

99 

100 return ret 

101 

102 

103class NullAuth(requests.auth.AuthBase): 

104 """A custom auth class which ensures requests does not override authorization 

105 headers with netrc file credentials if present. 

106 """ 

107 

108 def __call__(self, r): 

109 return r