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""" 

25 

26import logging 

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

28 

29import requests 

30 

31from looker_sdk.rtl import transport 

32 

33 

34class RequestsTransport(transport.Transport): 

35 """RequestsTransport implementation of Transport.""" 

36 

37 def __init__( 

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

39 ): 

40 self.settings = settings 

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

42 if settings.headers: 

43 headers.update(settings.headers) 

44 session.headers.update(headers) 

45 session.verify = settings.verify_ssl 

46 self.session = session 

47 self.logger = logging.getLogger(__name__) 

48 

49 @classmethod 

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

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

52 

53 def request( 

54 self, 

55 method: transport.HttpMethod, 

56 path: str, 

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

58 body: Optional[bytes] = None, 

59 authenticator: transport.TAuthenticator = None, 

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

61 ) -> transport.Response: 

62 

63 headers = {} 

64 timeout = self.settings.timeout 

65 if authenticator: 

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

67 if transport_options: 

68 if transport_options.get("headers"): 

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

70 if transport_options.get("timeout"): 

71 timeout = transport_options["timeout"] 

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

73 try: 

74 resp = self.session.request( 

75 method.name, 

76 path, 

77 auth=NullAuth(), 

78 params=query_params, 

79 data=body, 

80 headers=headers, 

81 timeout=timeout, 

82 ) 

83 except IOError as exc: 

84 ret = transport.Response( 

85 False, 

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

87 transport.ResponseMode.STRING, 

88 ) 

89 else: 

90 ret = transport.Response( 

91 resp.ok, 

92 resp.content, 

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

94 ) 

95 encoding = cast( 

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

97 ) 

98 if encoding: 

99 ret.encoding = encoding 

100 

101 return ret 

102 

103 

104class NullAuth(requests.auth.AuthBase): 

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

106 headers with netrc file credentials if present. 

107 """ 

108 

109 def __call__(self, r): 

110 return r