Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/cloud/secretmanager_v1beta1/services/secret_manager_service/transports/base.py: 74%

88 statements  

« prev     ^ index     » next       coverage.py v7.3.1, created at 2023-09-25 06:37 +0000

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

2# Copyright 2023 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 

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.iam.v1 import iam_policy_pb2 # type: ignore 

26from google.iam.v1 import policy_pb2 # type: ignore 

27from google.oauth2 import service_account # type: ignore 

28from google.protobuf import empty_pb2 # type: ignore 

29 

30from google.cloud.secretmanager_v1beta1 import gapic_version as package_version 

31from google.cloud.secretmanager_v1beta1.types import resources, service 

32 

33DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo( 

34 gapic_version=package_version.__version__ 

35) 

36 

37 

38class SecretManagerServiceTransport(abc.ABC): 

39 """Abstract transport class for SecretManagerService.""" 

40 

41 AUTH_SCOPES = ("https://www.googleapis.com/auth/cloud-platform",) 

42 

43 DEFAULT_HOST: str = "secretmanager.googleapis.com" 

44 

45 def __init__( 

46 self, 

47 *, 

48 host: str = DEFAULT_HOST, 

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

50 credentials_file: Optional[str] = None, 

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

52 quota_project_id: Optional[str] = None, 

53 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

54 always_use_jwt_access: Optional[bool] = False, 

55 api_audience: Optional[str] = None, 

56 **kwargs, 

57 ) -> None: 

58 """Instantiate the transport. 

59 

60 Args: 

61 host (Optional[str]): 

62 The hostname to connect to. 

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

64 authorization credentials to attach to requests. These 

65 credentials identify the application to the service; if none 

66 are specified, the client will attempt to ascertain the 

67 credentials from the environment. 

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

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

70 This argument is mutually exclusive with credentials. 

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

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

73 and quota. 

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

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

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

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

78 your own client library. 

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

80 be used for service account credentials. 

81 """ 

82 

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

84 

85 # Save the scopes. 

86 self._scopes = scopes 

87 

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

89 # defaults. 

90 if credentials and credentials_file: 

91 raise core_exceptions.DuplicateCredentialArgs( 

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

93 ) 

94 

95 if credentials_file is not None: 

96 credentials, _ = google.auth.load_credentials_from_file( 

97 credentials_file, **scopes_kwargs, quota_project_id=quota_project_id 

98 ) 

99 elif credentials is None: 

100 credentials, _ = google.auth.default( 

101 **scopes_kwargs, quota_project_id=quota_project_id 

102 ) 

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

104 if hasattr(credentials, "with_gdch_audience"): 

105 credentials = credentials.with_gdch_audience( 

106 api_audience if api_audience else host 

107 ) 

108 

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

110 if ( 

111 always_use_jwt_access 

112 and isinstance(credentials, service_account.Credentials) 

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

114 ): 

115 credentials = credentials.with_always_use_jwt_access(True) 

116 

117 # Save the credentials. 

118 self._credentials = credentials 

119 

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

121 if ":" not in host: 

122 host += ":443" 

123 self._host = host 

124 

125 def _prep_wrapped_messages(self, client_info): 

126 # Precompute the wrapped methods. 

127 self._wrapped_methods = { 

128 self.list_secrets: gapic_v1.method.wrap_method( 

129 self.list_secrets, 

130 default_timeout=60.0, 

131 client_info=client_info, 

132 ), 

133 self.create_secret: gapic_v1.method.wrap_method( 

134 self.create_secret, 

135 default_timeout=60.0, 

136 client_info=client_info, 

137 ), 

138 self.add_secret_version: gapic_v1.method.wrap_method( 

139 self.add_secret_version, 

140 default_timeout=60.0, 

141 client_info=client_info, 

142 ), 

143 self.get_secret: gapic_v1.method.wrap_method( 

144 self.get_secret, 

145 default_timeout=60.0, 

146 client_info=client_info, 

147 ), 

148 self.update_secret: gapic_v1.method.wrap_method( 

149 self.update_secret, 

150 default_timeout=60.0, 

151 client_info=client_info, 

152 ), 

153 self.delete_secret: gapic_v1.method.wrap_method( 

154 self.delete_secret, 

155 default_timeout=60.0, 

156 client_info=client_info, 

157 ), 

158 self.list_secret_versions: gapic_v1.method.wrap_method( 

159 self.list_secret_versions, 

160 default_timeout=60.0, 

161 client_info=client_info, 

162 ), 

163 self.get_secret_version: gapic_v1.method.wrap_method( 

164 self.get_secret_version, 

165 default_timeout=60.0, 

166 client_info=client_info, 

167 ), 

168 self.access_secret_version: gapic_v1.method.wrap_method( 

169 self.access_secret_version, 

170 default_retry=retries.Retry( 

171 initial=1.0, 

172 maximum=60.0, 

173 multiplier=1.3, 

174 predicate=retries.if_exception_type( 

175 core_exceptions.ServiceUnavailable, 

176 core_exceptions.Unknown, 

177 ), 

178 deadline=60.0, 

179 ), 

180 default_timeout=60.0, 

181 client_info=client_info, 

182 ), 

183 self.disable_secret_version: gapic_v1.method.wrap_method( 

184 self.disable_secret_version, 

185 default_timeout=60.0, 

186 client_info=client_info, 

187 ), 

188 self.enable_secret_version: gapic_v1.method.wrap_method( 

189 self.enable_secret_version, 

190 default_timeout=60.0, 

191 client_info=client_info, 

192 ), 

193 self.destroy_secret_version: gapic_v1.method.wrap_method( 

194 self.destroy_secret_version, 

195 default_timeout=60.0, 

196 client_info=client_info, 

197 ), 

198 self.set_iam_policy: gapic_v1.method.wrap_method( 

199 self.set_iam_policy, 

200 default_timeout=60.0, 

201 client_info=client_info, 

202 ), 

203 self.get_iam_policy: gapic_v1.method.wrap_method( 

204 self.get_iam_policy, 

205 default_timeout=60.0, 

206 client_info=client_info, 

207 ), 

208 self.test_iam_permissions: gapic_v1.method.wrap_method( 

209 self.test_iam_permissions, 

210 default_timeout=60.0, 

211 client_info=client_info, 

212 ), 

213 } 

214 

215 def close(self): 

216 """Closes resources associated with the transport. 

217 

218 .. warning:: 

219 Only call this method if the transport is NOT shared 

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

221 """ 

222 raise NotImplementedError() 

223 

224 @property 

225 def list_secrets( 

226 self, 

227 ) -> Callable[ 

228 [service.ListSecretsRequest], 

229 Union[service.ListSecretsResponse, Awaitable[service.ListSecretsResponse]], 

230 ]: 

231 raise NotImplementedError() 

232 

233 @property 

234 def create_secret( 

235 self, 

236 ) -> Callable[ 

237 [service.CreateSecretRequest], 

238 Union[resources.Secret, Awaitable[resources.Secret]], 

239 ]: 

240 raise NotImplementedError() 

241 

242 @property 

243 def add_secret_version( 

244 self, 

245 ) -> Callable[ 

246 [service.AddSecretVersionRequest], 

247 Union[resources.SecretVersion, Awaitable[resources.SecretVersion]], 

248 ]: 

249 raise NotImplementedError() 

250 

251 @property 

252 def get_secret( 

253 self, 

254 ) -> Callable[ 

255 [service.GetSecretRequest], Union[resources.Secret, Awaitable[resources.Secret]] 

256 ]: 

257 raise NotImplementedError() 

258 

259 @property 

260 def update_secret( 

261 self, 

262 ) -> Callable[ 

263 [service.UpdateSecretRequest], 

264 Union[resources.Secret, Awaitable[resources.Secret]], 

265 ]: 

266 raise NotImplementedError() 

267 

268 @property 

269 def delete_secret( 

270 self, 

271 ) -> Callable[ 

272 [service.DeleteSecretRequest], 

273 Union[empty_pb2.Empty, Awaitable[empty_pb2.Empty]], 

274 ]: 

275 raise NotImplementedError() 

276 

277 @property 

278 def list_secret_versions( 

279 self, 

280 ) -> Callable[ 

281 [service.ListSecretVersionsRequest], 

282 Union[ 

283 service.ListSecretVersionsResponse, 

284 Awaitable[service.ListSecretVersionsResponse], 

285 ], 

286 ]: 

287 raise NotImplementedError() 

288 

289 @property 

290 def get_secret_version( 

291 self, 

292 ) -> Callable[ 

293 [service.GetSecretVersionRequest], 

294 Union[resources.SecretVersion, Awaitable[resources.SecretVersion]], 

295 ]: 

296 raise NotImplementedError() 

297 

298 @property 

299 def access_secret_version( 

300 self, 

301 ) -> Callable[ 

302 [service.AccessSecretVersionRequest], 

303 Union[ 

304 service.AccessSecretVersionResponse, 

305 Awaitable[service.AccessSecretVersionResponse], 

306 ], 

307 ]: 

308 raise NotImplementedError() 

309 

310 @property 

311 def disable_secret_version( 

312 self, 

313 ) -> Callable[ 

314 [service.DisableSecretVersionRequest], 

315 Union[resources.SecretVersion, Awaitable[resources.SecretVersion]], 

316 ]: 

317 raise NotImplementedError() 

318 

319 @property 

320 def enable_secret_version( 

321 self, 

322 ) -> Callable[ 

323 [service.EnableSecretVersionRequest], 

324 Union[resources.SecretVersion, Awaitable[resources.SecretVersion]], 

325 ]: 

326 raise NotImplementedError() 

327 

328 @property 

329 def destroy_secret_version( 

330 self, 

331 ) -> Callable[ 

332 [service.DestroySecretVersionRequest], 

333 Union[resources.SecretVersion, Awaitable[resources.SecretVersion]], 

334 ]: 

335 raise NotImplementedError() 

336 

337 @property 

338 def set_iam_policy( 

339 self, 

340 ) -> Callable[ 

341 [iam_policy_pb2.SetIamPolicyRequest], 

342 Union[policy_pb2.Policy, Awaitable[policy_pb2.Policy]], 

343 ]: 

344 raise NotImplementedError() 

345 

346 @property 

347 def get_iam_policy( 

348 self, 

349 ) -> Callable[ 

350 [iam_policy_pb2.GetIamPolicyRequest], 

351 Union[policy_pb2.Policy, Awaitable[policy_pb2.Policy]], 

352 ]: 

353 raise NotImplementedError() 

354 

355 @property 

356 def test_iam_permissions( 

357 self, 

358 ) -> Callable[ 

359 [iam_policy_pb2.TestIamPermissionsRequest], 

360 Union[ 

361 iam_policy_pb2.TestIamPermissionsResponse, 

362 Awaitable[iam_policy_pb2.TestIamPermissionsResponse], 

363 ], 

364 ]: 

365 raise NotImplementedError() 

366 

367 @property 

368 def kind(self) -> str: 

369 raise NotImplementedError() 

370 

371 

372__all__ = ("SecretManagerServiceTransport",)