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 2024 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.iam.v1 import iam_policy_pb2 # type: ignore
22from google.iam.v1 import policy_pb2 # type: ignore
23from google.protobuf import empty_pb2 # type: ignore
24from google.protobuf import json_format
26from google.cloud.secretmanager_v1beta1.types import resources, service
28from .base import DEFAULT_CLIENT_INFO, SecretManagerServiceTransport
31class _BaseSecretManagerServiceRestTransport(SecretManagerServiceTransport):
32 """Base REST backend transport for SecretManagerService.
34 Note: This class is not meant to be used directly. Use its sync and
35 async sub-classes instead.
37 This class defines the same methods as the primary client, so the
38 primary client can load the underlying transport implementation
39 and call it.
41 It sends JSON representations of protocol buffers over HTTP/1.1
42 """
44 def __init__(
45 self,
46 *,
47 host: str = "secretmanager.googleapis.com",
48 credentials: Optional[Any] = None,
49 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
50 always_use_jwt_access: Optional[bool] = False,
51 url_scheme: str = "https",
52 api_audience: Optional[str] = None,
53 ) -> None:
54 """Instantiate the transport.
55 Args:
56 host (Optional[str]):
57 The hostname to connect to (default: 'secretmanager.googleapis.com').
58 credentials (Optional[Any]): The
59 authorization credentials to attach to requests. These
60 credentials identify the application to the service; if none
61 are specified, the client will attempt to ascertain the
62 credentials from the environment.
63 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
64 The client info used to send a user-agent string along with
65 API requests. If ``None``, then default info will be used.
66 Generally, you only need to set this if you are developing
67 your own client library.
68 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
69 be used for service account credentials.
70 url_scheme: the protocol scheme for the API endpoint. Normally
71 "https", but for testing or local servers,
72 "http" can be specified.
73 """
74 # Run the base constructor
75 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
76 if maybe_url_match is None:
77 raise ValueError(
78 f"Unexpected hostname structure: {host}"
79 ) # pragma: NO COVER
81 url_match_items = maybe_url_match.groupdict()
83 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
85 super().__init__(
86 host=host,
87 credentials=credentials,
88 client_info=client_info,
89 always_use_jwt_access=always_use_jwt_access,
90 api_audience=api_audience,
91 )
93 class _BaseAccessSecretVersion:
94 def __hash__(self): # pragma: NO COVER
95 return NotImplementedError("__hash__ must be implemented.")
97 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
99 @classmethod
100 def _get_unset_required_fields(cls, message_dict):
101 return {
102 k: v
103 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
104 if k not in message_dict
105 }
107 @staticmethod
108 def _get_http_options():
109 http_options: List[Dict[str, str]] = [
110 {
111 "method": "get",
112 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:access",
113 },
114 ]
115 return http_options
117 @staticmethod
118 def _get_transcoded_request(http_options, request):
119 pb_request = service.AccessSecretVersionRequest.pb(request)
120 transcoded_request = path_template.transcode(http_options, pb_request)
121 return transcoded_request
123 @staticmethod
124 def _get_query_params_json(transcoded_request):
125 query_params = json.loads(
126 json_format.MessageToJson(
127 transcoded_request["query_params"],
128 use_integers_for_enums=True,
129 )
130 )
131 query_params.update(
132 _BaseSecretManagerServiceRestTransport._BaseAccessSecretVersion._get_unset_required_fields(
133 query_params
134 )
135 )
137 query_params["$alt"] = "json;enum-encoding=int"
138 return query_params
140 class _BaseAddSecretVersion:
141 def __hash__(self): # pragma: NO COVER
142 return NotImplementedError("__hash__ must be implemented.")
144 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
146 @classmethod
147 def _get_unset_required_fields(cls, message_dict):
148 return {
149 k: v
150 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
151 if k not in message_dict
152 }
154 @staticmethod
155 def _get_http_options():
156 http_options: List[Dict[str, str]] = [
157 {
158 "method": "post",
159 "uri": "/v1beta1/{parent=projects/*/secrets/*}:addVersion",
160 "body": "*",
161 },
162 ]
163 return http_options
165 @staticmethod
166 def _get_transcoded_request(http_options, request):
167 pb_request = service.AddSecretVersionRequest.pb(request)
168 transcoded_request = path_template.transcode(http_options, pb_request)
169 return transcoded_request
171 @staticmethod
172 def _get_request_body_json(transcoded_request):
173 # Jsonify the request body
175 body = json_format.MessageToJson(
176 transcoded_request["body"], use_integers_for_enums=True
177 )
178 return body
180 @staticmethod
181 def _get_query_params_json(transcoded_request):
182 query_params = json.loads(
183 json_format.MessageToJson(
184 transcoded_request["query_params"],
185 use_integers_for_enums=True,
186 )
187 )
188 query_params.update(
189 _BaseSecretManagerServiceRestTransport._BaseAddSecretVersion._get_unset_required_fields(
190 query_params
191 )
192 )
194 query_params["$alt"] = "json;enum-encoding=int"
195 return query_params
197 class _BaseCreateSecret:
198 def __hash__(self): # pragma: NO COVER
199 return NotImplementedError("__hash__ must be implemented.")
201 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {
202 "secretId": "",
203 }
205 @classmethod
206 def _get_unset_required_fields(cls, message_dict):
207 return {
208 k: v
209 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
210 if k not in message_dict
211 }
213 @staticmethod
214 def _get_http_options():
215 http_options: List[Dict[str, str]] = [
216 {
217 "method": "post",
218 "uri": "/v1beta1/{parent=projects/*}/secrets",
219 "body": "secret",
220 },
221 ]
222 return http_options
224 @staticmethod
225 def _get_transcoded_request(http_options, request):
226 pb_request = service.CreateSecretRequest.pb(request)
227 transcoded_request = path_template.transcode(http_options, pb_request)
228 return transcoded_request
230 @staticmethod
231 def _get_request_body_json(transcoded_request):
232 # Jsonify the request body
234 body = json_format.MessageToJson(
235 transcoded_request["body"], use_integers_for_enums=True
236 )
237 return body
239 @staticmethod
240 def _get_query_params_json(transcoded_request):
241 query_params = json.loads(
242 json_format.MessageToJson(
243 transcoded_request["query_params"],
244 use_integers_for_enums=True,
245 )
246 )
247 query_params.update(
248 _BaseSecretManagerServiceRestTransport._BaseCreateSecret._get_unset_required_fields(
249 query_params
250 )
251 )
253 query_params["$alt"] = "json;enum-encoding=int"
254 return query_params
256 class _BaseDeleteSecret:
257 def __hash__(self): # pragma: NO COVER
258 return NotImplementedError("__hash__ must be implemented.")
260 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
262 @classmethod
263 def _get_unset_required_fields(cls, message_dict):
264 return {
265 k: v
266 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
267 if k not in message_dict
268 }
270 @staticmethod
271 def _get_http_options():
272 http_options: List[Dict[str, str]] = [
273 {
274 "method": "delete",
275 "uri": "/v1beta1/{name=projects/*/secrets/*}",
276 },
277 ]
278 return http_options
280 @staticmethod
281 def _get_transcoded_request(http_options, request):
282 pb_request = service.DeleteSecretRequest.pb(request)
283 transcoded_request = path_template.transcode(http_options, pb_request)
284 return transcoded_request
286 @staticmethod
287 def _get_query_params_json(transcoded_request):
288 query_params = json.loads(
289 json_format.MessageToJson(
290 transcoded_request["query_params"],
291 use_integers_for_enums=True,
292 )
293 )
294 query_params.update(
295 _BaseSecretManagerServiceRestTransport._BaseDeleteSecret._get_unset_required_fields(
296 query_params
297 )
298 )
300 query_params["$alt"] = "json;enum-encoding=int"
301 return query_params
303 class _BaseDestroySecretVersion:
304 def __hash__(self): # pragma: NO COVER
305 return NotImplementedError("__hash__ must be implemented.")
307 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
309 @classmethod
310 def _get_unset_required_fields(cls, message_dict):
311 return {
312 k: v
313 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
314 if k not in message_dict
315 }
317 @staticmethod
318 def _get_http_options():
319 http_options: List[Dict[str, str]] = [
320 {
321 "method": "post",
322 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:destroy",
323 "body": "*",
324 },
325 ]
326 return http_options
328 @staticmethod
329 def _get_transcoded_request(http_options, request):
330 pb_request = service.DestroySecretVersionRequest.pb(request)
331 transcoded_request = path_template.transcode(http_options, pb_request)
332 return transcoded_request
334 @staticmethod
335 def _get_request_body_json(transcoded_request):
336 # Jsonify the request body
338 body = json_format.MessageToJson(
339 transcoded_request["body"], use_integers_for_enums=True
340 )
341 return body
343 @staticmethod
344 def _get_query_params_json(transcoded_request):
345 query_params = json.loads(
346 json_format.MessageToJson(
347 transcoded_request["query_params"],
348 use_integers_for_enums=True,
349 )
350 )
351 query_params.update(
352 _BaseSecretManagerServiceRestTransport._BaseDestroySecretVersion._get_unset_required_fields(
353 query_params
354 )
355 )
357 query_params["$alt"] = "json;enum-encoding=int"
358 return query_params
360 class _BaseDisableSecretVersion:
361 def __hash__(self): # pragma: NO COVER
362 return NotImplementedError("__hash__ must be implemented.")
364 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
366 @classmethod
367 def _get_unset_required_fields(cls, message_dict):
368 return {
369 k: v
370 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
371 if k not in message_dict
372 }
374 @staticmethod
375 def _get_http_options():
376 http_options: List[Dict[str, str]] = [
377 {
378 "method": "post",
379 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:disable",
380 "body": "*",
381 },
382 ]
383 return http_options
385 @staticmethod
386 def _get_transcoded_request(http_options, request):
387 pb_request = service.DisableSecretVersionRequest.pb(request)
388 transcoded_request = path_template.transcode(http_options, pb_request)
389 return transcoded_request
391 @staticmethod
392 def _get_request_body_json(transcoded_request):
393 # Jsonify the request body
395 body = json_format.MessageToJson(
396 transcoded_request["body"], use_integers_for_enums=True
397 )
398 return body
400 @staticmethod
401 def _get_query_params_json(transcoded_request):
402 query_params = json.loads(
403 json_format.MessageToJson(
404 transcoded_request["query_params"],
405 use_integers_for_enums=True,
406 )
407 )
408 query_params.update(
409 _BaseSecretManagerServiceRestTransport._BaseDisableSecretVersion._get_unset_required_fields(
410 query_params
411 )
412 )
414 query_params["$alt"] = "json;enum-encoding=int"
415 return query_params
417 class _BaseEnableSecretVersion:
418 def __hash__(self): # pragma: NO COVER
419 return NotImplementedError("__hash__ must be implemented.")
421 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
423 @classmethod
424 def _get_unset_required_fields(cls, message_dict):
425 return {
426 k: v
427 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
428 if k not in message_dict
429 }
431 @staticmethod
432 def _get_http_options():
433 http_options: List[Dict[str, str]] = [
434 {
435 "method": "post",
436 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}:enable",
437 "body": "*",
438 },
439 ]
440 return http_options
442 @staticmethod
443 def _get_transcoded_request(http_options, request):
444 pb_request = service.EnableSecretVersionRequest.pb(request)
445 transcoded_request = path_template.transcode(http_options, pb_request)
446 return transcoded_request
448 @staticmethod
449 def _get_request_body_json(transcoded_request):
450 # Jsonify the request body
452 body = json_format.MessageToJson(
453 transcoded_request["body"], use_integers_for_enums=True
454 )
455 return body
457 @staticmethod
458 def _get_query_params_json(transcoded_request):
459 query_params = json.loads(
460 json_format.MessageToJson(
461 transcoded_request["query_params"],
462 use_integers_for_enums=True,
463 )
464 )
465 query_params.update(
466 _BaseSecretManagerServiceRestTransport._BaseEnableSecretVersion._get_unset_required_fields(
467 query_params
468 )
469 )
471 query_params["$alt"] = "json;enum-encoding=int"
472 return query_params
474 class _BaseGetIamPolicy:
475 def __hash__(self): # pragma: NO COVER
476 return NotImplementedError("__hash__ must be implemented.")
478 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
480 @classmethod
481 def _get_unset_required_fields(cls, message_dict):
482 return {
483 k: v
484 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
485 if k not in message_dict
486 }
488 @staticmethod
489 def _get_http_options():
490 http_options: List[Dict[str, str]] = [
491 {
492 "method": "get",
493 "uri": "/v1beta1/{resource=projects/*/secrets/*}:getIamPolicy",
494 },
495 ]
496 return http_options
498 @staticmethod
499 def _get_transcoded_request(http_options, request):
500 pb_request = request
501 transcoded_request = path_template.transcode(http_options, pb_request)
502 return transcoded_request
504 @staticmethod
505 def _get_query_params_json(transcoded_request):
506 query_params = json.loads(
507 json_format.MessageToJson(
508 transcoded_request["query_params"],
509 use_integers_for_enums=True,
510 )
511 )
512 query_params.update(
513 _BaseSecretManagerServiceRestTransport._BaseGetIamPolicy._get_unset_required_fields(
514 query_params
515 )
516 )
518 query_params["$alt"] = "json;enum-encoding=int"
519 return query_params
521 class _BaseGetSecret:
522 def __hash__(self): # pragma: NO COVER
523 return NotImplementedError("__hash__ must be implemented.")
525 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
527 @classmethod
528 def _get_unset_required_fields(cls, message_dict):
529 return {
530 k: v
531 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
532 if k not in message_dict
533 }
535 @staticmethod
536 def _get_http_options():
537 http_options: List[Dict[str, str]] = [
538 {
539 "method": "get",
540 "uri": "/v1beta1/{name=projects/*/secrets/*}",
541 },
542 ]
543 return http_options
545 @staticmethod
546 def _get_transcoded_request(http_options, request):
547 pb_request = service.GetSecretRequest.pb(request)
548 transcoded_request = path_template.transcode(http_options, pb_request)
549 return transcoded_request
551 @staticmethod
552 def _get_query_params_json(transcoded_request):
553 query_params = json.loads(
554 json_format.MessageToJson(
555 transcoded_request["query_params"],
556 use_integers_for_enums=True,
557 )
558 )
559 query_params.update(
560 _BaseSecretManagerServiceRestTransport._BaseGetSecret._get_unset_required_fields(
561 query_params
562 )
563 )
565 query_params["$alt"] = "json;enum-encoding=int"
566 return query_params
568 class _BaseGetSecretVersion:
569 def __hash__(self): # pragma: NO COVER
570 return NotImplementedError("__hash__ must be implemented.")
572 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
574 @classmethod
575 def _get_unset_required_fields(cls, message_dict):
576 return {
577 k: v
578 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
579 if k not in message_dict
580 }
582 @staticmethod
583 def _get_http_options():
584 http_options: List[Dict[str, str]] = [
585 {
586 "method": "get",
587 "uri": "/v1beta1/{name=projects/*/secrets/*/versions/*}",
588 },
589 ]
590 return http_options
592 @staticmethod
593 def _get_transcoded_request(http_options, request):
594 pb_request = service.GetSecretVersionRequest.pb(request)
595 transcoded_request = path_template.transcode(http_options, pb_request)
596 return transcoded_request
598 @staticmethod
599 def _get_query_params_json(transcoded_request):
600 query_params = json.loads(
601 json_format.MessageToJson(
602 transcoded_request["query_params"],
603 use_integers_for_enums=True,
604 )
605 )
606 query_params.update(
607 _BaseSecretManagerServiceRestTransport._BaseGetSecretVersion._get_unset_required_fields(
608 query_params
609 )
610 )
612 query_params["$alt"] = "json;enum-encoding=int"
613 return query_params
615 class _BaseListSecrets:
616 def __hash__(self): # pragma: NO COVER
617 return NotImplementedError("__hash__ must be implemented.")
619 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
621 @classmethod
622 def _get_unset_required_fields(cls, message_dict):
623 return {
624 k: v
625 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
626 if k not in message_dict
627 }
629 @staticmethod
630 def _get_http_options():
631 http_options: List[Dict[str, str]] = [
632 {
633 "method": "get",
634 "uri": "/v1beta1/{parent=projects/*}/secrets",
635 },
636 ]
637 return http_options
639 @staticmethod
640 def _get_transcoded_request(http_options, request):
641 pb_request = service.ListSecretsRequest.pb(request)
642 transcoded_request = path_template.transcode(http_options, pb_request)
643 return transcoded_request
645 @staticmethod
646 def _get_query_params_json(transcoded_request):
647 query_params = json.loads(
648 json_format.MessageToJson(
649 transcoded_request["query_params"],
650 use_integers_for_enums=True,
651 )
652 )
653 query_params.update(
654 _BaseSecretManagerServiceRestTransport._BaseListSecrets._get_unset_required_fields(
655 query_params
656 )
657 )
659 query_params["$alt"] = "json;enum-encoding=int"
660 return query_params
662 class _BaseListSecretVersions:
663 def __hash__(self): # pragma: NO COVER
664 return NotImplementedError("__hash__ must be implemented.")
666 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
668 @classmethod
669 def _get_unset_required_fields(cls, message_dict):
670 return {
671 k: v
672 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
673 if k not in message_dict
674 }
676 @staticmethod
677 def _get_http_options():
678 http_options: List[Dict[str, str]] = [
679 {
680 "method": "get",
681 "uri": "/v1beta1/{parent=projects/*/secrets/*}/versions",
682 },
683 ]
684 return http_options
686 @staticmethod
687 def _get_transcoded_request(http_options, request):
688 pb_request = service.ListSecretVersionsRequest.pb(request)
689 transcoded_request = path_template.transcode(http_options, pb_request)
690 return transcoded_request
692 @staticmethod
693 def _get_query_params_json(transcoded_request):
694 query_params = json.loads(
695 json_format.MessageToJson(
696 transcoded_request["query_params"],
697 use_integers_for_enums=True,
698 )
699 )
700 query_params.update(
701 _BaseSecretManagerServiceRestTransport._BaseListSecretVersions._get_unset_required_fields(
702 query_params
703 )
704 )
706 query_params["$alt"] = "json;enum-encoding=int"
707 return query_params
709 class _BaseSetIamPolicy:
710 def __hash__(self): # pragma: NO COVER
711 return NotImplementedError("__hash__ must be implemented.")
713 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
715 @classmethod
716 def _get_unset_required_fields(cls, message_dict):
717 return {
718 k: v
719 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
720 if k not in message_dict
721 }
723 @staticmethod
724 def _get_http_options():
725 http_options: List[Dict[str, str]] = [
726 {
727 "method": "post",
728 "uri": "/v1beta1/{resource=projects/*/secrets/*}:setIamPolicy",
729 "body": "*",
730 },
731 ]
732 return http_options
734 @staticmethod
735 def _get_transcoded_request(http_options, request):
736 pb_request = request
737 transcoded_request = path_template.transcode(http_options, pb_request)
738 return transcoded_request
740 @staticmethod
741 def _get_request_body_json(transcoded_request):
742 # Jsonify the request body
744 body = json_format.MessageToJson(
745 transcoded_request["body"], use_integers_for_enums=True
746 )
747 return body
749 @staticmethod
750 def _get_query_params_json(transcoded_request):
751 query_params = json.loads(
752 json_format.MessageToJson(
753 transcoded_request["query_params"],
754 use_integers_for_enums=True,
755 )
756 )
757 query_params.update(
758 _BaseSecretManagerServiceRestTransport._BaseSetIamPolicy._get_unset_required_fields(
759 query_params
760 )
761 )
763 query_params["$alt"] = "json;enum-encoding=int"
764 return query_params
766 class _BaseTestIamPermissions:
767 def __hash__(self): # pragma: NO COVER
768 return NotImplementedError("__hash__ must be implemented.")
770 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
772 @classmethod
773 def _get_unset_required_fields(cls, message_dict):
774 return {
775 k: v
776 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
777 if k not in message_dict
778 }
780 @staticmethod
781 def _get_http_options():
782 http_options: List[Dict[str, str]] = [
783 {
784 "method": "post",
785 "uri": "/v1beta1/{resource=projects/*/secrets/*}:testIamPermissions",
786 "body": "*",
787 },
788 ]
789 return http_options
791 @staticmethod
792 def _get_transcoded_request(http_options, request):
793 pb_request = request
794 transcoded_request = path_template.transcode(http_options, pb_request)
795 return transcoded_request
797 @staticmethod
798 def _get_request_body_json(transcoded_request):
799 # Jsonify the request body
801 body = json_format.MessageToJson(
802 transcoded_request["body"], use_integers_for_enums=True
803 )
804 return body
806 @staticmethod
807 def _get_query_params_json(transcoded_request):
808 query_params = json.loads(
809 json_format.MessageToJson(
810 transcoded_request["query_params"],
811 use_integers_for_enums=True,
812 )
813 )
814 query_params.update(
815 _BaseSecretManagerServiceRestTransport._BaseTestIamPermissions._get_unset_required_fields(
816 query_params
817 )
818 )
820 query_params["$alt"] = "json;enum-encoding=int"
821 return query_params
823 class _BaseUpdateSecret:
824 def __hash__(self): # pragma: NO COVER
825 return NotImplementedError("__hash__ must be implemented.")
827 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {
828 "updateMask": {},
829 }
831 @classmethod
832 def _get_unset_required_fields(cls, message_dict):
833 return {
834 k: v
835 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
836 if k not in message_dict
837 }
839 @staticmethod
840 def _get_http_options():
841 http_options: List[Dict[str, str]] = [
842 {
843 "method": "patch",
844 "uri": "/v1beta1/{secret.name=projects/*/secrets/*}",
845 "body": "secret",
846 },
847 ]
848 return http_options
850 @staticmethod
851 def _get_transcoded_request(http_options, request):
852 pb_request = service.UpdateSecretRequest.pb(request)
853 transcoded_request = path_template.transcode(http_options, pb_request)
854 return transcoded_request
856 @staticmethod
857 def _get_request_body_json(transcoded_request):
858 # Jsonify the request body
860 body = json_format.MessageToJson(
861 transcoded_request["body"], use_integers_for_enums=True
862 )
863 return body
865 @staticmethod
866 def _get_query_params_json(transcoded_request):
867 query_params = json.loads(
868 json_format.MessageToJson(
869 transcoded_request["query_params"],
870 use_integers_for_enums=True,
871 )
872 )
873 query_params.update(
874 _BaseSecretManagerServiceRestTransport._BaseUpdateSecret._get_unset_required_fields(
875 query_params
876 )
877 )
879 query_params["$alt"] = "json;enum-encoding=int"
880 return query_params
883__all__ = ("_BaseSecretManagerServiceRestTransport",)