Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/secretmanager_v1beta1/services/secret_manager_service/transports/rest_base.py: 67%

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

2# Copyright 2026 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 json # type: ignore 

17import re 

18from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union 

19 

20import google.iam.v1.iam_policy_pb2 as iam_policy_pb2 # type: ignore 

21import google.iam.v1.policy_pb2 as policy_pb2 # type: ignore 

22import google.protobuf.empty_pb2 as empty_pb2 # type: ignore 

23from google.api_core import gapic_v1, path_template 

24from google.cloud.location import locations_pb2 # type: ignore 

25from google.protobuf import json_format 

26 

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

28 

29from .base import DEFAULT_CLIENT_INFO, SecretManagerServiceTransport 

30 

31 

32class _BaseSecretManagerServiceRestTransport(SecretManagerServiceTransport): 

33 """Base REST backend transport for SecretManagerService. 

34 

35 Note: This class is not meant to be used directly. Use its sync and 

36 async sub-classes instead. 

37 

38 This class defines the same methods as the primary client, so the 

39 primary client can load the underlying transport implementation 

40 and call it. 

41 

42 It sends JSON representations of protocol buffers over HTTP/1.1 

43 """ 

44 

45 def __init__( 

46 self, 

47 *, 

48 host: str = "secretmanager.googleapis.com", 

49 credentials: Optional[Any] = None, 

50 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

51 always_use_jwt_access: Optional[bool] = False, 

52 url_scheme: str = "https", 

53 api_audience: Optional[str] = None, 

54 ) -> None: 

55 """Instantiate the transport. 

56 Args: 

57 host (Optional[str]): 

58 The hostname to connect to (default: 'secretmanager.googleapis.com'). 

59 credentials (Optional[Any]): The 

60 authorization credentials to attach to requests. These 

61 credentials identify the application to the service; if none 

62 are specified, the client will attempt to ascertain the 

63 credentials from the environment. 

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

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

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

67 Generally, you only need to set this if you are developing 

68 your own client library. 

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

70 be used for service account credentials. 

71 url_scheme: the protocol scheme for the API endpoint. Normally 

72 "https", but for testing or local servers, 

73 "http" can be specified. 

74 """ 

75 # Run the base constructor 

76 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host) 

77 if maybe_url_match is None: 

78 raise ValueError( 

79 f"Unexpected hostname structure: {host}" 

80 ) # pragma: NO COVER 

81 

82 url_match_items = maybe_url_match.groupdict() 

83 

84 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host 

85 

86 super().__init__( 

87 host=host, 

88 credentials=credentials, 

89 client_info=client_info, 

90 always_use_jwt_access=always_use_jwt_access, 

91 api_audience=api_audience, 

92 ) 

93 

94 class _BaseAccessSecretVersion: 

95 def __hash__(self): # pragma: NO COVER 

96 return NotImplementedError("__hash__ must be implemented.") 

97 

98 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

99 

100 @staticmethod 

101 def _get_http_options(): 

102 http_options: List[Dict[str, str]] = [ 

103 { 

104 "method": "get", 

105 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:access", 

106 }, 

107 ] 

108 return http_options 

109 

110 class _BaseAddSecretVersion: 

111 def __hash__(self): # pragma: NO COVER 

112 return NotImplementedError("__hash__ must be implemented.") 

113 

114 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

115 

116 @staticmethod 

117 def _get_http_options(): 

118 http_options: List[Dict[str, str]] = [ 

119 { 

120 "method": "post", 

121 "uri": "/v1beta1/{parent=projects/*/secrets/*}:addVersion", 

122 "body": "*", 

123 }, 

124 ] 

125 return http_options 

126 

127 class _BaseCreateSecret: 

128 def __hash__(self): # pragma: NO COVER 

129 return NotImplementedError("__hash__ must be implemented.") 

130 

131 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = { 

132 "secretId": "", 

133 } 

134 

135 @staticmethod 

136 def _get_http_options(): 

137 http_options: List[Dict[str, str]] = [ 

138 { 

139 "method": "post", 

140 "uri": "/v1beta1/{parent=projects/*}/secrets", 

141 "body": "secret", 

142 }, 

143 ] 

144 return http_options 

145 

146 class _BaseDeleteSecret: 

147 def __hash__(self): # pragma: NO COVER 

148 return NotImplementedError("__hash__ must be implemented.") 

149 

150 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

151 

152 @staticmethod 

153 def _get_http_options(): 

154 http_options: List[Dict[str, str]] = [ 

155 { 

156 "method": "delete", 

157 "uri": "/v1beta1/{name=projects/*/secrets/*}", 

158 }, 

159 ] 

160 return http_options 

161 

162 class _BaseDestroySecretVersion: 

163 def __hash__(self): # pragma: NO COVER 

164 return NotImplementedError("__hash__ must be implemented.") 

165 

166 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

167 

168 @staticmethod 

169 def _get_http_options(): 

170 http_options: List[Dict[str, str]] = [ 

171 { 

172 "method": "post", 

173 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:destroy", 

174 "body": "*", 

175 }, 

176 ] 

177 return http_options 

178 

179 class _BaseDisableSecretVersion: 

180 def __hash__(self): # pragma: NO COVER 

181 return NotImplementedError("__hash__ must be implemented.") 

182 

183 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

184 

185 @staticmethod 

186 def _get_http_options(): 

187 http_options: List[Dict[str, str]] = [ 

188 { 

189 "method": "post", 

190 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:disable", 

191 "body": "*", 

192 }, 

193 ] 

194 return http_options 

195 

196 class _BaseEnableSecretVersion: 

197 def __hash__(self): # pragma: NO COVER 

198 return NotImplementedError("__hash__ must be implemented.") 

199 

200 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

201 

202 @staticmethod 

203 def _get_http_options(): 

204 http_options: List[Dict[str, str]] = [ 

205 { 

206 "method": "post", 

207 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:enable", 

208 "body": "*", 

209 }, 

210 ] 

211 return http_options 

212 

213 class _BaseGetIamPolicy: 

214 def __hash__(self): # pragma: NO COVER 

215 return NotImplementedError("__hash__ must be implemented.") 

216 

217 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

218 

219 @staticmethod 

220 def _get_http_options(): 

221 http_options: List[Dict[str, str]] = [ 

222 { 

223 "method": "get", 

224 "uri": "/v1beta1/{resource=projects/*/secrets/*}:getIamPolicy", 

225 }, 

226 ] 

227 return http_options 

228 

229 class _BaseGetSecret: 

230 def __hash__(self): # pragma: NO COVER 

231 return NotImplementedError("__hash__ must be implemented.") 

232 

233 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

234 

235 @staticmethod 

236 def _get_http_options(): 

237 http_options: List[Dict[str, str]] = [ 

238 { 

239 "method": "get", 

240 "uri": "/v1beta1/{name=projects/*/secrets/*}", 

241 }, 

242 ] 

243 return http_options 

244 

245 class _BaseGetSecretVersion: 

246 def __hash__(self): # pragma: NO COVER 

247 return NotImplementedError("__hash__ must be implemented.") 

248 

249 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

250 

251 @staticmethod 

252 def _get_http_options(): 

253 http_options: List[Dict[str, str]] = [ 

254 { 

255 "method": "get", 

256 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}", 

257 }, 

258 ] 

259 return http_options 

260 

261 class _BaseListSecrets: 

262 def __hash__(self): # pragma: NO COVER 

263 return NotImplementedError("__hash__ must be implemented.") 

264 

265 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

266 

267 @staticmethod 

268 def _get_http_options(): 

269 http_options: List[Dict[str, str]] = [ 

270 { 

271 "method": "get", 

272 "uri": "/v1beta1/{parent=projects/*}/secrets", 

273 }, 

274 ] 

275 return http_options 

276 

277 class _BaseListSecretVersions: 

278 def __hash__(self): # pragma: NO COVER 

279 return NotImplementedError("__hash__ must be implemented.") 

280 

281 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

282 

283 @staticmethod 

284 def _get_http_options(): 

285 http_options: List[Dict[str, str]] = [ 

286 { 

287 "method": "get", 

288 "uri": "/v1beta1/{parent=projects/*/secrets/*}/versions", 

289 }, 

290 ] 

291 return http_options 

292 

293 class _BaseSetIamPolicy: 

294 def __hash__(self): # pragma: NO COVER 

295 return NotImplementedError("__hash__ must be implemented.") 

296 

297 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

298 

299 @staticmethod 

300 def _get_http_options(): 

301 http_options: List[Dict[str, str]] = [ 

302 { 

303 "method": "post", 

304 "uri": "/v1beta1/{resource=projects/*/secrets/*}:setIamPolicy", 

305 "body": "*", 

306 }, 

307 ] 

308 return http_options 

309 

310 class _BaseTestIamPermissions: 

311 def __hash__(self): # pragma: NO COVER 

312 return NotImplementedError("__hash__ must be implemented.") 

313 

314 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

315 

316 @staticmethod 

317 def _get_http_options(): 

318 http_options: List[Dict[str, str]] = [ 

319 { 

320 "method": "post", 

321 "uri": "/v1beta1/{resource=projects/*/secrets/*}:testIamPermissions", 

322 "body": "*", 

323 }, 

324 ] 

325 return http_options 

326 

327 class _BaseUpdateSecret: 

328 def __hash__(self): # pragma: NO COVER 

329 return NotImplementedError("__hash__ must be implemented.") 

330 

331 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = { 

332 "updateMask": {}, 

333 } 

334 

335 @staticmethod 

336 def _get_http_options(): 

337 http_options: List[Dict[str, str]] = [ 

338 { 

339 "method": "patch", 

340 "uri": "/v1beta1/{secret.name=projects/*/secrets/*}", 

341 "body": "secret", 

342 }, 

343 ] 

344 return http_options 

345 

346 class _BaseGetLocation: 

347 def __hash__(self): # pragma: NO COVER 

348 return NotImplementedError("__hash__ must be implemented.") 

349 

350 @staticmethod 

351 def _get_http_options(): 

352 http_options: List[Dict[str, str]] = [ 

353 { 

354 "method": "get", 

355 "uri": "/v1beta1/{name=projects/*/locations/*}", 

356 }, 

357 ] 

358 return http_options 

359 

360 class _BaseListLocations: 

361 def __hash__(self): # pragma: NO COVER 

362 return NotImplementedError("__hash__ must be implemented.") 

363 

364 @staticmethod 

365 def _get_http_options(): 

366 http_options: List[Dict[str, str]] = [ 

367 { 

368 "method": "get", 

369 "uri": "/v1beta1/{name=projects/*}/locations", 

370 }, 

371 ] 

372 return http_options 

373 

374 

375__all__ = ("_BaseSecretManagerServiceRestTransport",)