Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# -*- coding: utf-8 -*-
2# Copyright 2025 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
20from google.api_core import gapic_v1, path_template
21from google.cloud.location import locations_pb2 # type: ignore
22from google.iam.v1 import iam_policy_pb2 # type: ignore
23from google.iam.v1 import policy_pb2 # type: ignore
24from google.protobuf import empty_pb2 # type: ignore
25from google.protobuf import json_format
27from google.cloud.secretmanager_v1beta1.types import resources, service
29from .base import DEFAULT_CLIENT_INFO, SecretManagerServiceTransport
32class _BaseSecretManagerServiceRestTransport(SecretManagerServiceTransport):
33 """Base REST backend transport for SecretManagerService.
35 Note: This class is not meant to be used directly. Use its sync and
36 async sub-classes instead.
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.
42 It sends JSON representations of protocol buffers over HTTP/1.1
43 """
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
82 url_match_items = maybe_url_match.groupdict()
84 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
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 )
94 class _BaseAccessSecretVersion:
95 def __hash__(self): # pragma: NO COVER
96 return NotImplementedError("__hash__ must be implemented.")
98 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
100 @classmethod
101 def _get_unset_required_fields(cls, message_dict):
102 return {
103 k: v
104 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
105 if k not in message_dict
106 }
108 @staticmethod
109 def _get_http_options():
110 http_options: List[Dict[str, str]] = [
111 {
112 "method": "get",
113 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:access",
114 },
115 ]
116 return http_options
118 @staticmethod
119 def _get_transcoded_request(http_options, request):
120 pb_request = service.AccessSecretVersionRequest.pb(request)
121 transcoded_request = path_template.transcode(http_options, pb_request)
122 return transcoded_request
124 @staticmethod
125 def _get_query_params_json(transcoded_request):
126 query_params = json.loads(
127 json_format.MessageToJson(
128 transcoded_request["query_params"],
129 use_integers_for_enums=True,
130 )
131 )
132 query_params.update(
133 _BaseSecretManagerServiceRestTransport._BaseAccessSecretVersion._get_unset_required_fields(
134 query_params
135 )
136 )
138 query_params["$alt"] = "json;enum-encoding=int"
139 return query_params
141 class _BaseAddSecretVersion:
142 def __hash__(self): # pragma: NO COVER
143 return NotImplementedError("__hash__ must be implemented.")
145 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
147 @classmethod
148 def _get_unset_required_fields(cls, message_dict):
149 return {
150 k: v
151 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
152 if k not in message_dict
153 }
155 @staticmethod
156 def _get_http_options():
157 http_options: List[Dict[str, str]] = [
158 {
159 "method": "post",
160 "uri": "/v1beta1/{parent=projects/*/secrets/*}:addVersion",
161 "body": "*",
162 },
163 ]
164 return http_options
166 @staticmethod
167 def _get_transcoded_request(http_options, request):
168 pb_request = service.AddSecretVersionRequest.pb(request)
169 transcoded_request = path_template.transcode(http_options, pb_request)
170 return transcoded_request
172 @staticmethod
173 def _get_request_body_json(transcoded_request):
174 # Jsonify the request body
176 body = json_format.MessageToJson(
177 transcoded_request["body"], use_integers_for_enums=True
178 )
179 return body
181 @staticmethod
182 def _get_query_params_json(transcoded_request):
183 query_params = json.loads(
184 json_format.MessageToJson(
185 transcoded_request["query_params"],
186 use_integers_for_enums=True,
187 )
188 )
189 query_params.update(
190 _BaseSecretManagerServiceRestTransport._BaseAddSecretVersion._get_unset_required_fields(
191 query_params
192 )
193 )
195 query_params["$alt"] = "json;enum-encoding=int"
196 return query_params
198 class _BaseCreateSecret:
199 def __hash__(self): # pragma: NO COVER
200 return NotImplementedError("__hash__ must be implemented.")
202 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {
203 "secretId": "",
204 }
206 @classmethod
207 def _get_unset_required_fields(cls, message_dict):
208 return {
209 k: v
210 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
211 if k not in message_dict
212 }
214 @staticmethod
215 def _get_http_options():
216 http_options: List[Dict[str, str]] = [
217 {
218 "method": "post",
219 "uri": "/v1beta1/{parent=projects/*}/secrets",
220 "body": "secret",
221 },
222 ]
223 return http_options
225 @staticmethod
226 def _get_transcoded_request(http_options, request):
227 pb_request = service.CreateSecretRequest.pb(request)
228 transcoded_request = path_template.transcode(http_options, pb_request)
229 return transcoded_request
231 @staticmethod
232 def _get_request_body_json(transcoded_request):
233 # Jsonify the request body
235 body = json_format.MessageToJson(
236 transcoded_request["body"], use_integers_for_enums=True
237 )
238 return body
240 @staticmethod
241 def _get_query_params_json(transcoded_request):
242 query_params = json.loads(
243 json_format.MessageToJson(
244 transcoded_request["query_params"],
245 use_integers_for_enums=True,
246 )
247 )
248 query_params.update(
249 _BaseSecretManagerServiceRestTransport._BaseCreateSecret._get_unset_required_fields(
250 query_params
251 )
252 )
254 query_params["$alt"] = "json;enum-encoding=int"
255 return query_params
257 class _BaseDeleteSecret:
258 def __hash__(self): # pragma: NO COVER
259 return NotImplementedError("__hash__ must be implemented.")
261 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
263 @classmethod
264 def _get_unset_required_fields(cls, message_dict):
265 return {
266 k: v
267 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
268 if k not in message_dict
269 }
271 @staticmethod
272 def _get_http_options():
273 http_options: List[Dict[str, str]] = [
274 {
275 "method": "delete",
276 "uri": "/v1beta1/{name=projects/*/secrets/*}",
277 },
278 ]
279 return http_options
281 @staticmethod
282 def _get_transcoded_request(http_options, request):
283 pb_request = service.DeleteSecretRequest.pb(request)
284 transcoded_request = path_template.transcode(http_options, pb_request)
285 return transcoded_request
287 @staticmethod
288 def _get_query_params_json(transcoded_request):
289 query_params = json.loads(
290 json_format.MessageToJson(
291 transcoded_request["query_params"],
292 use_integers_for_enums=True,
293 )
294 )
295 query_params.update(
296 _BaseSecretManagerServiceRestTransport._BaseDeleteSecret._get_unset_required_fields(
297 query_params
298 )
299 )
301 query_params["$alt"] = "json;enum-encoding=int"
302 return query_params
304 class _BaseDestroySecretVersion:
305 def __hash__(self): # pragma: NO COVER
306 return NotImplementedError("__hash__ must be implemented.")
308 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
310 @classmethod
311 def _get_unset_required_fields(cls, message_dict):
312 return {
313 k: v
314 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
315 if k not in message_dict
316 }
318 @staticmethod
319 def _get_http_options():
320 http_options: List[Dict[str, str]] = [
321 {
322 "method": "post",
323 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:destroy",
324 "body": "*",
325 },
326 ]
327 return http_options
329 @staticmethod
330 def _get_transcoded_request(http_options, request):
331 pb_request = service.DestroySecretVersionRequest.pb(request)
332 transcoded_request = path_template.transcode(http_options, pb_request)
333 return transcoded_request
335 @staticmethod
336 def _get_request_body_json(transcoded_request):
337 # Jsonify the request body
339 body = json_format.MessageToJson(
340 transcoded_request["body"], use_integers_for_enums=True
341 )
342 return body
344 @staticmethod
345 def _get_query_params_json(transcoded_request):
346 query_params = json.loads(
347 json_format.MessageToJson(
348 transcoded_request["query_params"],
349 use_integers_for_enums=True,
350 )
351 )
352 query_params.update(
353 _BaseSecretManagerServiceRestTransport._BaseDestroySecretVersion._get_unset_required_fields(
354 query_params
355 )
356 )
358 query_params["$alt"] = "json;enum-encoding=int"
359 return query_params
361 class _BaseDisableSecretVersion:
362 def __hash__(self): # pragma: NO COVER
363 return NotImplementedError("__hash__ must be implemented.")
365 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
367 @classmethod
368 def _get_unset_required_fields(cls, message_dict):
369 return {
370 k: v
371 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
372 if k not in message_dict
373 }
375 @staticmethod
376 def _get_http_options():
377 http_options: List[Dict[str, str]] = [
378 {
379 "method": "post",
380 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:disable",
381 "body": "*",
382 },
383 ]
384 return http_options
386 @staticmethod
387 def _get_transcoded_request(http_options, request):
388 pb_request = service.DisableSecretVersionRequest.pb(request)
389 transcoded_request = path_template.transcode(http_options, pb_request)
390 return transcoded_request
392 @staticmethod
393 def _get_request_body_json(transcoded_request):
394 # Jsonify the request body
396 body = json_format.MessageToJson(
397 transcoded_request["body"], use_integers_for_enums=True
398 )
399 return body
401 @staticmethod
402 def _get_query_params_json(transcoded_request):
403 query_params = json.loads(
404 json_format.MessageToJson(
405 transcoded_request["query_params"],
406 use_integers_for_enums=True,
407 )
408 )
409 query_params.update(
410 _BaseSecretManagerServiceRestTransport._BaseDisableSecretVersion._get_unset_required_fields(
411 query_params
412 )
413 )
415 query_params["$alt"] = "json;enum-encoding=int"
416 return query_params
418 class _BaseEnableSecretVersion:
419 def __hash__(self): # pragma: NO COVER
420 return NotImplementedError("__hash__ must be implemented.")
422 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
424 @classmethod
425 def _get_unset_required_fields(cls, message_dict):
426 return {
427 k: v
428 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
429 if k not in message_dict
430 }
432 @staticmethod
433 def _get_http_options():
434 http_options: List[Dict[str, str]] = [
435 {
436 "method": "post",
437 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:enable",
438 "body": "*",
439 },
440 ]
441 return http_options
443 @staticmethod
444 def _get_transcoded_request(http_options, request):
445 pb_request = service.EnableSecretVersionRequest.pb(request)
446 transcoded_request = path_template.transcode(http_options, pb_request)
447 return transcoded_request
449 @staticmethod
450 def _get_request_body_json(transcoded_request):
451 # Jsonify the request body
453 body = json_format.MessageToJson(
454 transcoded_request["body"], use_integers_for_enums=True
455 )
456 return body
458 @staticmethod
459 def _get_query_params_json(transcoded_request):
460 query_params = json.loads(
461 json_format.MessageToJson(
462 transcoded_request["query_params"],
463 use_integers_for_enums=True,
464 )
465 )
466 query_params.update(
467 _BaseSecretManagerServiceRestTransport._BaseEnableSecretVersion._get_unset_required_fields(
468 query_params
469 )
470 )
472 query_params["$alt"] = "json;enum-encoding=int"
473 return query_params
475 class _BaseGetIamPolicy:
476 def __hash__(self): # pragma: NO COVER
477 return NotImplementedError("__hash__ must be implemented.")
479 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
481 @classmethod
482 def _get_unset_required_fields(cls, message_dict):
483 return {
484 k: v
485 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
486 if k not in message_dict
487 }
489 @staticmethod
490 def _get_http_options():
491 http_options: List[Dict[str, str]] = [
492 {
493 "method": "get",
494 "uri": "/v1beta1/{resource=projects/*/secrets/*}:getIamPolicy",
495 },
496 ]
497 return http_options
499 @staticmethod
500 def _get_transcoded_request(http_options, request):
501 pb_request = request
502 transcoded_request = path_template.transcode(http_options, pb_request)
503 return transcoded_request
505 @staticmethod
506 def _get_query_params_json(transcoded_request):
507 query_params = json.loads(
508 json_format.MessageToJson(
509 transcoded_request["query_params"],
510 use_integers_for_enums=True,
511 )
512 )
513 query_params.update(
514 _BaseSecretManagerServiceRestTransport._BaseGetIamPolicy._get_unset_required_fields(
515 query_params
516 )
517 )
519 query_params["$alt"] = "json;enum-encoding=int"
520 return query_params
522 class _BaseGetSecret:
523 def __hash__(self): # pragma: NO COVER
524 return NotImplementedError("__hash__ must be implemented.")
526 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
528 @classmethod
529 def _get_unset_required_fields(cls, message_dict):
530 return {
531 k: v
532 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
533 if k not in message_dict
534 }
536 @staticmethod
537 def _get_http_options():
538 http_options: List[Dict[str, str]] = [
539 {
540 "method": "get",
541 "uri": "/v1beta1/{name=projects/*/secrets/*}",
542 },
543 ]
544 return http_options
546 @staticmethod
547 def _get_transcoded_request(http_options, request):
548 pb_request = service.GetSecretRequest.pb(request)
549 transcoded_request = path_template.transcode(http_options, pb_request)
550 return transcoded_request
552 @staticmethod
553 def _get_query_params_json(transcoded_request):
554 query_params = json.loads(
555 json_format.MessageToJson(
556 transcoded_request["query_params"],
557 use_integers_for_enums=True,
558 )
559 )
560 query_params.update(
561 _BaseSecretManagerServiceRestTransport._BaseGetSecret._get_unset_required_fields(
562 query_params
563 )
564 )
566 query_params["$alt"] = "json;enum-encoding=int"
567 return query_params
569 class _BaseGetSecretVersion:
570 def __hash__(self): # pragma: NO COVER
571 return NotImplementedError("__hash__ must be implemented.")
573 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
575 @classmethod
576 def _get_unset_required_fields(cls, message_dict):
577 return {
578 k: v
579 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
580 if k not in message_dict
581 }
583 @staticmethod
584 def _get_http_options():
585 http_options: List[Dict[str, str]] = [
586 {
587 "method": "get",
588 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}",
589 },
590 ]
591 return http_options
593 @staticmethod
594 def _get_transcoded_request(http_options, request):
595 pb_request = service.GetSecretVersionRequest.pb(request)
596 transcoded_request = path_template.transcode(http_options, pb_request)
597 return transcoded_request
599 @staticmethod
600 def _get_query_params_json(transcoded_request):
601 query_params = json.loads(
602 json_format.MessageToJson(
603 transcoded_request["query_params"],
604 use_integers_for_enums=True,
605 )
606 )
607 query_params.update(
608 _BaseSecretManagerServiceRestTransport._BaseGetSecretVersion._get_unset_required_fields(
609 query_params
610 )
611 )
613 query_params["$alt"] = "json;enum-encoding=int"
614 return query_params
616 class _BaseListSecrets:
617 def __hash__(self): # pragma: NO COVER
618 return NotImplementedError("__hash__ must be implemented.")
620 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
622 @classmethod
623 def _get_unset_required_fields(cls, message_dict):
624 return {
625 k: v
626 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
627 if k not in message_dict
628 }
630 @staticmethod
631 def _get_http_options():
632 http_options: List[Dict[str, str]] = [
633 {
634 "method": "get",
635 "uri": "/v1beta1/{parent=projects/*}/secrets",
636 },
637 ]
638 return http_options
640 @staticmethod
641 def _get_transcoded_request(http_options, request):
642 pb_request = service.ListSecretsRequest.pb(request)
643 transcoded_request = path_template.transcode(http_options, pb_request)
644 return transcoded_request
646 @staticmethod
647 def _get_query_params_json(transcoded_request):
648 query_params = json.loads(
649 json_format.MessageToJson(
650 transcoded_request["query_params"],
651 use_integers_for_enums=True,
652 )
653 )
654 query_params.update(
655 _BaseSecretManagerServiceRestTransport._BaseListSecrets._get_unset_required_fields(
656 query_params
657 )
658 )
660 query_params["$alt"] = "json;enum-encoding=int"
661 return query_params
663 class _BaseListSecretVersions:
664 def __hash__(self): # pragma: NO COVER
665 return NotImplementedError("__hash__ must be implemented.")
667 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
669 @classmethod
670 def _get_unset_required_fields(cls, message_dict):
671 return {
672 k: v
673 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
674 if k not in message_dict
675 }
677 @staticmethod
678 def _get_http_options():
679 http_options: List[Dict[str, str]] = [
680 {
681 "method": "get",
682 "uri": "/v1beta1/{parent=projects/*/secrets/*}/versions",
683 },
684 ]
685 return http_options
687 @staticmethod
688 def _get_transcoded_request(http_options, request):
689 pb_request = service.ListSecretVersionsRequest.pb(request)
690 transcoded_request = path_template.transcode(http_options, pb_request)
691 return transcoded_request
693 @staticmethod
694 def _get_query_params_json(transcoded_request):
695 query_params = json.loads(
696 json_format.MessageToJson(
697 transcoded_request["query_params"],
698 use_integers_for_enums=True,
699 )
700 )
701 query_params.update(
702 _BaseSecretManagerServiceRestTransport._BaseListSecretVersions._get_unset_required_fields(
703 query_params
704 )
705 )
707 query_params["$alt"] = "json;enum-encoding=int"
708 return query_params
710 class _BaseSetIamPolicy:
711 def __hash__(self): # pragma: NO COVER
712 return NotImplementedError("__hash__ must be implemented.")
714 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
716 @classmethod
717 def _get_unset_required_fields(cls, message_dict):
718 return {
719 k: v
720 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
721 if k not in message_dict
722 }
724 @staticmethod
725 def _get_http_options():
726 http_options: List[Dict[str, str]] = [
727 {
728 "method": "post",
729 "uri": "/v1beta1/{resource=projects/*/secrets/*}:setIamPolicy",
730 "body": "*",
731 },
732 ]
733 return http_options
735 @staticmethod
736 def _get_transcoded_request(http_options, request):
737 pb_request = request
738 transcoded_request = path_template.transcode(http_options, pb_request)
739 return transcoded_request
741 @staticmethod
742 def _get_request_body_json(transcoded_request):
743 # Jsonify the request body
745 body = json_format.MessageToJson(
746 transcoded_request["body"], use_integers_for_enums=True
747 )
748 return body
750 @staticmethod
751 def _get_query_params_json(transcoded_request):
752 query_params = json.loads(
753 json_format.MessageToJson(
754 transcoded_request["query_params"],
755 use_integers_for_enums=True,
756 )
757 )
758 query_params.update(
759 _BaseSecretManagerServiceRestTransport._BaseSetIamPolicy._get_unset_required_fields(
760 query_params
761 )
762 )
764 query_params["$alt"] = "json;enum-encoding=int"
765 return query_params
767 class _BaseTestIamPermissions:
768 def __hash__(self): # pragma: NO COVER
769 return NotImplementedError("__hash__ must be implemented.")
771 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
773 @classmethod
774 def _get_unset_required_fields(cls, message_dict):
775 return {
776 k: v
777 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
778 if k not in message_dict
779 }
781 @staticmethod
782 def _get_http_options():
783 http_options: List[Dict[str, str]] = [
784 {
785 "method": "post",
786 "uri": "/v1beta1/{resource=projects/*/secrets/*}:testIamPermissions",
787 "body": "*",
788 },
789 ]
790 return http_options
792 @staticmethod
793 def _get_transcoded_request(http_options, request):
794 pb_request = request
795 transcoded_request = path_template.transcode(http_options, pb_request)
796 return transcoded_request
798 @staticmethod
799 def _get_request_body_json(transcoded_request):
800 # Jsonify the request body
802 body = json_format.MessageToJson(
803 transcoded_request["body"], use_integers_for_enums=True
804 )
805 return body
807 @staticmethod
808 def _get_query_params_json(transcoded_request):
809 query_params = json.loads(
810 json_format.MessageToJson(
811 transcoded_request["query_params"],
812 use_integers_for_enums=True,
813 )
814 )
815 query_params.update(
816 _BaseSecretManagerServiceRestTransport._BaseTestIamPermissions._get_unset_required_fields(
817 query_params
818 )
819 )
821 query_params["$alt"] = "json;enum-encoding=int"
822 return query_params
824 class _BaseUpdateSecret:
825 def __hash__(self): # pragma: NO COVER
826 return NotImplementedError("__hash__ must be implemented.")
828 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {
829 "updateMask": {},
830 }
832 @classmethod
833 def _get_unset_required_fields(cls, message_dict):
834 return {
835 k: v
836 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
837 if k not in message_dict
838 }
840 @staticmethod
841 def _get_http_options():
842 http_options: List[Dict[str, str]] = [
843 {
844 "method": "patch",
845 "uri": "/v1beta1/{secret.name=projects/*/secrets/*}",
846 "body": "secret",
847 },
848 ]
849 return http_options
851 @staticmethod
852 def _get_transcoded_request(http_options, request):
853 pb_request = service.UpdateSecretRequest.pb(request)
854 transcoded_request = path_template.transcode(http_options, pb_request)
855 return transcoded_request
857 @staticmethod
858 def _get_request_body_json(transcoded_request):
859 # Jsonify the request body
861 body = json_format.MessageToJson(
862 transcoded_request["body"], use_integers_for_enums=True
863 )
864 return body
866 @staticmethod
867 def _get_query_params_json(transcoded_request):
868 query_params = json.loads(
869 json_format.MessageToJson(
870 transcoded_request["query_params"],
871 use_integers_for_enums=True,
872 )
873 )
874 query_params.update(
875 _BaseSecretManagerServiceRestTransport._BaseUpdateSecret._get_unset_required_fields(
876 query_params
877 )
878 )
880 query_params["$alt"] = "json;enum-encoding=int"
881 return query_params
883 class _BaseGetLocation:
884 def __hash__(self): # pragma: NO COVER
885 return NotImplementedError("__hash__ must be implemented.")
887 @staticmethod
888 def _get_http_options():
889 http_options: List[Dict[str, str]] = [
890 {
891 "method": "get",
892 "uri": "/v1beta1/{name=projects/*/locations/*}",
893 },
894 ]
895 return http_options
897 @staticmethod
898 def _get_transcoded_request(http_options, request):
899 request_kwargs = json_format.MessageToDict(request)
900 transcoded_request = path_template.transcode(http_options, **request_kwargs)
901 return transcoded_request
903 @staticmethod
904 def _get_query_params_json(transcoded_request):
905 query_params = json.loads(json.dumps(transcoded_request["query_params"]))
906 return query_params
908 class _BaseListLocations:
909 def __hash__(self): # pragma: NO COVER
910 return NotImplementedError("__hash__ must be implemented.")
912 @staticmethod
913 def _get_http_options():
914 http_options: List[Dict[str, str]] = [
915 {
916 "method": "get",
917 "uri": "/v1beta1/{name=projects/*}/locations",
918 },
919 ]
920 return http_options
922 @staticmethod
923 def _get_transcoded_request(http_options, request):
924 request_kwargs = json_format.MessageToDict(request)
925 transcoded_request = path_template.transcode(http_options, **request_kwargs)
926 return transcoded_request
928 @staticmethod
929 def _get_query_params_json(transcoded_request):
930 query_params = json.loads(json.dumps(transcoded_request["query_params"]))
931 return query_params
934__all__ = ("_BaseSecretManagerServiceRestTransport",)