Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/cloud/resourcemanager_v3/services/tag_holds/transports/base.py: 57%

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

2# Copyright 2022 Google LLC 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15# 

16import abc 

17from typing import Awaitable, Callable, Dict, Optional, Sequence, Union 

18 

19import google.api_core 

20from google.api_core import exceptions as core_exceptions 

21from google.api_core import gapic_v1, operations_v1 

22from google.api_core import retry as retries 

23import google.auth # type: ignore 

24from google.auth import credentials as ga_credentials # type: ignore 

25from google.longrunning import operations_pb2 # type: ignore 

26from google.oauth2 import service_account # type: ignore 

27 

28from google.cloud.resourcemanager_v3 import gapic_version as package_version 

29from google.cloud.resourcemanager_v3.types import tag_holds 

30 

31DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo( 

32 gapic_version=package_version.__version__ 

33) 

34 

35 

36class TagHoldsTransport(abc.ABC): 

37 """Abstract transport class for TagHolds.""" 

38 

39 AUTH_SCOPES = ( 

40 "https://www.googleapis.com/auth/cloud-platform", 

41 "https://www.googleapis.com/auth/cloud-platform.read-only", 

42 ) 

43 

44 DEFAULT_HOST: str = "cloudresourcemanager.googleapis.com" 

45 

46 def __init__( 

47 self, 

48 *, 

49 host: str = DEFAULT_HOST, 

50 credentials: Optional[ga_credentials.Credentials] = None, 

51 credentials_file: Optional[str] = None, 

52 scopes: Optional[Sequence[str]] = None, 

53 quota_project_id: Optional[str] = None, 

54 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

55 always_use_jwt_access: Optional[bool] = False, 

56 api_audience: Optional[str] = None, 

57 **kwargs, 

58 ) -> None: 

59 """Instantiate the transport. 

60 

61 Args: 

62 host (Optional[str]): 

63 The hostname to connect to. 

64 credentials (Optional[google.auth.credentials.Credentials]): The 

65 authorization credentials to attach to requests. These 

66 credentials identify the application to the service; if none 

67 are specified, the client will attempt to ascertain the 

68 credentials from the environment. 

69 credentials_file (Optional[str]): A file with credentials that can 

70 be loaded with :func:`google.auth.load_credentials_from_file`. 

71 This argument is mutually exclusive with credentials. 

72 scopes (Optional[Sequence[str]]): A list of scopes. 

73 quota_project_id (Optional[str]): An optional project to use for billing 

74 and quota. 

75 client_info (google.api_core.gapic_v1.client_info.ClientInfo): 

76 The client info used to send a user-agent string along with 

77 API requests. If ``None``, then default info will be used. 

78 Generally, you only need to set this if you're developing 

79 your own client library. 

80 always_use_jwt_access (Optional[bool]): Whether self signed JWT should 

81 be used for service account credentials. 

82 """ 

83 

84 scopes_kwargs = {"scopes": scopes, "default_scopes": self.AUTH_SCOPES} 

85 

86 # Save the scopes. 

87 self._scopes = scopes 

88 

89 # If no credentials are provided, then determine the appropriate 

90 # defaults. 

91 if credentials and credentials_file: 

92 raise core_exceptions.DuplicateCredentialArgs( 

93 "'credentials_file' and 'credentials' are mutually exclusive" 

94 ) 

95 

96 if credentials_file is not None: 

97 credentials, _ = google.auth.load_credentials_from_file( 

98 credentials_file, **scopes_kwargs, quota_project_id=quota_project_id 

99 ) 

100 elif credentials is None: 

101 credentials, _ = google.auth.default( 

102 **scopes_kwargs, quota_project_id=quota_project_id 

103 ) 

104 # Don't apply audience if the credentials file passed from user. 

105 if hasattr(credentials, "with_gdch_audience"): 

106 credentials = credentials.with_gdch_audience( 

107 api_audience if api_audience else host 

108 ) 

109 

110 # If the credentials are service account credentials, then always try to use self signed JWT. 

111 if ( 

112 always_use_jwt_access 

113 and isinstance(credentials, service_account.Credentials) 

114 and hasattr(service_account.Credentials, "with_always_use_jwt_access") 

115 ): 

116 credentials = credentials.with_always_use_jwt_access(True) 

117 

118 # Save the credentials. 

119 self._credentials = credentials 

120 

121 # Save the hostname. Default to port 443 (HTTPS) if none is specified. 

122 if ":" not in host: 

123 host += ":443" 

124 self._host = host 

125 

126 def _prep_wrapped_messages(self, client_info): 

127 # Precompute the wrapped methods. 

128 self._wrapped_methods = { 

129 self.create_tag_hold: gapic_v1.method.wrap_method( 

130 self.create_tag_hold, 

131 default_timeout=None, 

132 client_info=client_info, 

133 ), 

134 self.delete_tag_hold: gapic_v1.method.wrap_method( 

135 self.delete_tag_hold, 

136 default_timeout=None, 

137 client_info=client_info, 

138 ), 

139 self.list_tag_holds: gapic_v1.method.wrap_method( 

140 self.list_tag_holds, 

141 default_timeout=None, 

142 client_info=client_info, 

143 ), 

144 } 

145 

146 def close(self): 

147 """Closes resources associated with the transport. 

148 

149 .. warning:: 

150 Only call this method if the transport is NOT shared 

151 with other clients - this may cause errors in other clients! 

152 """ 

153 raise NotImplementedError() 

154 

155 @property 

156 def operations_client(self): 

157 """Return the client designed to process long-running operations.""" 

158 raise NotImplementedError() 

159 

160 @property 

161 def create_tag_hold( 

162 self, 

163 ) -> Callable[ 

164 [tag_holds.CreateTagHoldRequest], 

165 Union[operations_pb2.Operation, Awaitable[operations_pb2.Operation]], 

166 ]: 

167 raise NotImplementedError() 

168 

169 @property 

170 def delete_tag_hold( 

171 self, 

172 ) -> Callable[ 

173 [tag_holds.DeleteTagHoldRequest], 

174 Union[operations_pb2.Operation, Awaitable[operations_pb2.Operation]], 

175 ]: 

176 raise NotImplementedError() 

177 

178 @property 

179 def list_tag_holds( 

180 self, 

181 ) -> Callable[ 

182 [tag_holds.ListTagHoldsRequest], 

183 Union[ 

184 tag_holds.ListTagHoldsResponse, Awaitable[tag_holds.ListTagHoldsResponse] 

185 ], 

186 ]: 

187 raise NotImplementedError() 

188 

189 @property 

190 def get_operation( 

191 self, 

192 ) -> Callable[ 

193 [operations_pb2.GetOperationRequest], 

194 Union[operations_pb2.Operation, Awaitable[operations_pb2.Operation]], 

195 ]: 

196 raise NotImplementedError() 

197 

198 @property 

199 def kind(self) -> str: 

200 raise NotImplementedError() 

201 

202 

203__all__ = ("TagHoldsTransport",)