Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/botocore/model.py: 54%
434 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:51 +0000
1# Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2#
3# Licensed under the Apache License, Version 2.0 (the "License"). You
4# may not use this file except in compliance with the License. A copy of
5# the License is located at
6#
7# http://aws.amazon.com/apache2.0/
8#
9# or in the "license" file accompanying this file. This file is
10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11# ANY KIND, either express or implied. See the License for the specific
12# language governing permissions and limitations under the License.
13"""Abstractions to interact with service models."""
14from collections import defaultdict
15from typing import NamedTuple, Union
17from botocore.compat import OrderedDict
18from botocore.exceptions import (
19 MissingServiceIdError,
20 UndefinedModelAttributeError,
21)
22from botocore.utils import CachedProperty, hyphenize_service_id, instance_cache
24NOT_SET = object()
27class NoShapeFoundError(Exception):
28 pass
31class InvalidShapeError(Exception):
32 pass
35class OperationNotFoundError(Exception):
36 pass
39class InvalidShapeReferenceError(Exception):
40 pass
43class ServiceId(str):
44 def hyphenize(self):
45 return hyphenize_service_id(self)
48class Shape:
49 """Object representing a shape from the service model."""
51 # To simplify serialization logic, all shape params that are
52 # related to serialization are moved from the top level hash into
53 # a 'serialization' hash. This list below contains the names of all
54 # the attributes that should be moved.
55 SERIALIZED_ATTRS = [
56 'locationName',
57 'queryName',
58 'flattened',
59 'location',
60 'payload',
61 'streaming',
62 'timestampFormat',
63 'xmlNamespace',
64 'resultWrapper',
65 'xmlAttribute',
66 'eventstream',
67 'event',
68 'eventheader',
69 'eventpayload',
70 'jsonvalue',
71 'timestampFormat',
72 'hostLabel',
73 ]
74 METADATA_ATTRS = [
75 'required',
76 'min',
77 'max',
78 'pattern',
79 'sensitive',
80 'enum',
81 'idempotencyToken',
82 'error',
83 'exception',
84 'endpointdiscoveryid',
85 'retryable',
86 'document',
87 'union',
88 'contextParam',
89 'clientContextParams',
90 'requiresLength',
91 ]
92 MAP_TYPE = OrderedDict
94 def __init__(self, shape_name, shape_model, shape_resolver=None):
95 """
97 :type shape_name: string
98 :param shape_name: The name of the shape.
100 :type shape_model: dict
101 :param shape_model: The shape model. This would be the value
102 associated with the key in the "shapes" dict of the
103 service model (i.e ``model['shapes'][shape_name]``)
105 :type shape_resolver: botocore.model.ShapeResolver
106 :param shape_resolver: A shape resolver object. This is used to
107 resolve references to other shapes. For scalar shape types
108 (string, integer, boolean, etc.), this argument is not
109 required. If a shape_resolver is not provided for a complex
110 type, then a ``ValueError`` will be raised when an attempt
111 to resolve a shape is made.
113 """
114 self.name = shape_name
115 self.type_name = shape_model['type']
116 self.documentation = shape_model.get('documentation', '')
117 self._shape_model = shape_model
118 if shape_resolver is None:
119 # If a shape_resolver is not provided, we create an object
120 # that will throw errors if you attempt to resolve
121 # a shape. This is actually ok for scalar shapes
122 # because they don't need to resolve shapes and shouldn't
123 # be required to provide an object they won't use.
124 shape_resolver = UnresolvableShapeMap()
125 self._shape_resolver = shape_resolver
126 self._cache = {}
128 @CachedProperty
129 def serialization(self):
130 """Serialization information about the shape.
132 This contains information that may be needed for input serialization
133 or response parsing. This can include:
135 * name
136 * queryName
137 * flattened
138 * location
139 * payload
140 * streaming
141 * xmlNamespace
142 * resultWrapper
143 * xmlAttribute
144 * jsonvalue
145 * timestampFormat
147 :rtype: dict
148 :return: Serialization information about the shape.
150 """
151 model = self._shape_model
152 serialization = {}
153 for attr in self.SERIALIZED_ATTRS:
154 if attr in self._shape_model:
155 serialization[attr] = model[attr]
156 # For consistency, locationName is renamed to just 'name'.
157 if 'locationName' in serialization:
158 serialization['name'] = serialization.pop('locationName')
159 return serialization
161 @CachedProperty
162 def metadata(self):
163 """Metadata about the shape.
165 This requires optional information about the shape, including:
167 * min
168 * max
169 * pattern
170 * enum
171 * sensitive
172 * required
173 * idempotencyToken
174 * document
175 * union
176 * contextParam
177 * clientContextParams
178 * requiresLength
180 :rtype: dict
181 :return: Metadata about the shape.
183 """
184 model = self._shape_model
185 metadata = {}
186 for attr in self.METADATA_ATTRS:
187 if attr in self._shape_model:
188 metadata[attr] = model[attr]
189 return metadata
191 @CachedProperty
192 def required_members(self):
193 """A list of members that are required.
195 A structure shape can define members that are required.
196 This value will return a list of required members. If there
197 are no required members an empty list is returned.
199 """
200 return self.metadata.get('required', [])
202 def _resolve_shape_ref(self, shape_ref):
203 return self._shape_resolver.resolve_shape_ref(shape_ref)
205 def __repr__(self):
206 return f"<{self.__class__.__name__}({self.name})>"
208 @property
209 def event_stream_name(self):
210 return None
213class StructureShape(Shape):
214 @CachedProperty
215 def members(self):
216 members = self._shape_model.get('members', self.MAP_TYPE())
217 # The members dict looks like:
218 # 'members': {
219 # 'MemberName': {'shape': 'shapeName'},
220 # 'MemberName2': {'shape': 'shapeName'},
221 # }
222 # We return a dict of member name to Shape object.
223 shape_members = self.MAP_TYPE()
224 for name, shape_ref in members.items():
225 shape_members[name] = self._resolve_shape_ref(shape_ref)
226 return shape_members
228 @CachedProperty
229 def event_stream_name(self):
230 for member_name, member in self.members.items():
231 if member.serialization.get('eventstream'):
232 return member_name
233 return None
235 @CachedProperty
236 def error_code(self):
237 if not self.metadata.get('exception', False):
238 return None
239 error_metadata = self.metadata.get("error", {})
240 code = error_metadata.get("code")
241 if code:
242 return code
243 # Use the exception name if there is no explicit code modeled
244 return self.name
246 @CachedProperty
247 def is_document_type(self):
248 return self.metadata.get('document', False)
250 @CachedProperty
251 def is_tagged_union(self):
252 return self.metadata.get('union', False)
255class ListShape(Shape):
256 @CachedProperty
257 def member(self):
258 return self._resolve_shape_ref(self._shape_model['member'])
261class MapShape(Shape):
262 @CachedProperty
263 def key(self):
264 return self._resolve_shape_ref(self._shape_model['key'])
266 @CachedProperty
267 def value(self):
268 return self._resolve_shape_ref(self._shape_model['value'])
271class StringShape(Shape):
272 @CachedProperty
273 def enum(self):
274 return self.metadata.get('enum', [])
277class StaticContextParameter(NamedTuple):
278 name: str
279 value: Union[bool, str]
282class ContextParameter(NamedTuple):
283 name: str
284 member_name: str
287class ClientContextParameter(NamedTuple):
288 name: str
289 type: str
290 documentation: str
293class ServiceModel:
294 """
296 :ivar service_description: The parsed service description dictionary.
298 """
300 def __init__(self, service_description, service_name=None):
301 """
303 :type service_description: dict
304 :param service_description: The service description model. This value
305 is obtained from a botocore.loader.Loader, or from directly loading
306 the file yourself::
308 service_description = json.load(
309 open('/path/to/service-description-model.json'))
310 model = ServiceModel(service_description)
312 :type service_name: str
313 :param service_name: The name of the service. Normally this is
314 the endpoint prefix defined in the service_description. However,
315 you can override this value to provide a more convenient name.
316 This is done in a few places in botocore (ses instead of email,
317 emr instead of elasticmapreduce). If this value is not provided,
318 it will default to the endpointPrefix defined in the model.
320 """
321 self._service_description = service_description
322 # We want clients to be able to access metadata directly.
323 self.metadata = service_description.get('metadata', {})
324 self._shape_resolver = ShapeResolver(
325 service_description.get('shapes', {})
326 )
327 self._signature_version = NOT_SET
328 self._service_name = service_name
329 self._instance_cache = {}
331 def shape_for(self, shape_name, member_traits=None):
332 return self._shape_resolver.get_shape_by_name(
333 shape_name, member_traits
334 )
336 def shape_for_error_code(self, error_code):
337 return self._error_code_cache.get(error_code, None)
339 @CachedProperty
340 def _error_code_cache(self):
341 error_code_cache = {}
342 for error_shape in self.error_shapes:
343 code = error_shape.error_code
344 error_code_cache[code] = error_shape
345 return error_code_cache
347 def resolve_shape_ref(self, shape_ref):
348 return self._shape_resolver.resolve_shape_ref(shape_ref)
350 @CachedProperty
351 def shape_names(self):
352 return list(self._service_description.get('shapes', {}))
354 @CachedProperty
355 def error_shapes(self):
356 error_shapes = []
357 for shape_name in self.shape_names:
358 error_shape = self.shape_for(shape_name)
359 if error_shape.metadata.get('exception', False):
360 error_shapes.append(error_shape)
361 return error_shapes
363 @instance_cache
364 def operation_model(self, operation_name):
365 try:
366 model = self._service_description['operations'][operation_name]
367 except KeyError:
368 raise OperationNotFoundError(operation_name)
369 return OperationModel(model, self, operation_name)
371 @CachedProperty
372 def documentation(self):
373 return self._service_description.get('documentation', '')
375 @CachedProperty
376 def operation_names(self):
377 return list(self._service_description.get('operations', []))
379 @CachedProperty
380 def service_name(self):
381 """The name of the service.
383 This defaults to the endpointPrefix defined in the service model.
384 However, this value can be overriden when a ``ServiceModel`` is
385 created. If a service_name was not provided when the ``ServiceModel``
386 was created and if there is no endpointPrefix defined in the
387 service model, then an ``UndefinedModelAttributeError`` exception
388 will be raised.
390 """
391 if self._service_name is not None:
392 return self._service_name
393 else:
394 return self.endpoint_prefix
396 @CachedProperty
397 def service_id(self):
398 try:
399 return ServiceId(self._get_metadata_property('serviceId'))
400 except UndefinedModelAttributeError:
401 raise MissingServiceIdError(service_name=self._service_name)
403 @CachedProperty
404 def signing_name(self):
405 """The name to use when computing signatures.
407 If the model does not define a signing name, this
408 value will be the endpoint prefix defined in the model.
409 """
410 signing_name = self.metadata.get('signingName')
411 if signing_name is None:
412 signing_name = self.endpoint_prefix
413 return signing_name
415 @CachedProperty
416 def api_version(self):
417 return self._get_metadata_property('apiVersion')
419 @CachedProperty
420 def protocol(self):
421 return self._get_metadata_property('protocol')
423 @CachedProperty
424 def endpoint_prefix(self):
425 return self._get_metadata_property('endpointPrefix')
427 @CachedProperty
428 def endpoint_discovery_operation(self):
429 for operation in self.operation_names:
430 model = self.operation_model(operation)
431 if model.is_endpoint_discovery_operation:
432 return model
434 @CachedProperty
435 def endpoint_discovery_required(self):
436 for operation in self.operation_names:
437 model = self.operation_model(operation)
438 if (
439 model.endpoint_discovery is not None
440 and model.endpoint_discovery.get('required')
441 ):
442 return True
443 return False
445 @CachedProperty
446 def client_context_parameters(self):
447 params = self._service_description.get('clientContextParams', {})
448 return [
449 ClientContextParameter(
450 name=param_name,
451 type=param_val['type'],
452 documentation=param_val['documentation'],
453 )
454 for param_name, param_val in params.items()
455 ]
457 def _get_metadata_property(self, name):
458 try:
459 return self.metadata[name]
460 except KeyError:
461 raise UndefinedModelAttributeError(
462 f'"{name}" not defined in the metadata of the model: {self}'
463 )
465 # Signature version is one of the rare properties
466 # that can be modified so a CachedProperty is not used here.
468 @property
469 def signature_version(self):
470 if self._signature_version is NOT_SET:
471 signature_version = self.metadata.get('signatureVersion')
472 self._signature_version = signature_version
473 return self._signature_version
475 @signature_version.setter
476 def signature_version(self, value):
477 self._signature_version = value
479 def __repr__(self):
480 return f'{self.__class__.__name__}({self.service_name})'
483class OperationModel:
484 def __init__(self, operation_model, service_model, name=None):
485 """
487 :type operation_model: dict
488 :param operation_model: The operation model. This comes from the
489 service model, and is the value associated with the operation
490 name in the service model (i.e ``model['operations'][op_name]``).
492 :type service_model: botocore.model.ServiceModel
493 :param service_model: The service model associated with the operation.
495 :type name: string
496 :param name: The operation name. This is the operation name exposed to
497 the users of this model. This can potentially be different from
498 the "wire_name", which is the operation name that *must* by
499 provided over the wire. For example, given::
501 "CreateCloudFrontOriginAccessIdentity":{
502 "name":"CreateCloudFrontOriginAccessIdentity2014_11_06",
503 ...
504 }
506 The ``name`` would be ``CreateCloudFrontOriginAccessIdentity``,
507 but the ``self.wire_name`` would be
508 ``CreateCloudFrontOriginAccessIdentity2014_11_06``, which is the
509 value we must send in the corresponding HTTP request.
511 """
512 self._operation_model = operation_model
513 self._service_model = service_model
514 self._api_name = name
515 # Clients can access '.name' to get the operation name
516 # and '.metadata' to get the top level metdata of the service.
517 self._wire_name = operation_model.get('name')
518 self.metadata = service_model.metadata
519 self.http = operation_model.get('http', {})
521 @CachedProperty
522 def name(self):
523 if self._api_name is not None:
524 return self._api_name
525 else:
526 return self.wire_name
528 @property
529 def wire_name(self):
530 """The wire name of the operation.
532 In many situations this is the same value as the
533 ``name``, value, but in some services, the operation name
534 exposed to the user is different from the operation name
535 we send across the wire (e.g cloudfront).
537 Any serialization code should use ``wire_name``.
539 """
540 return self._operation_model.get('name')
542 @property
543 def service_model(self):
544 return self._service_model
546 @CachedProperty
547 def documentation(self):
548 return self._operation_model.get('documentation', '')
550 @CachedProperty
551 def deprecated(self):
552 return self._operation_model.get('deprecated', False)
554 @CachedProperty
555 def endpoint_discovery(self):
556 # Explicit None default. An empty dictionary for this trait means it is
557 # enabled but not required to be used.
558 return self._operation_model.get('endpointdiscovery', None)
560 @CachedProperty
561 def is_endpoint_discovery_operation(self):
562 return self._operation_model.get('endpointoperation', False)
564 @CachedProperty
565 def input_shape(self):
566 if 'input' not in self._operation_model:
567 # Some operations do not accept any input and do not define an
568 # input shape.
569 return None
570 return self._service_model.resolve_shape_ref(
571 self._operation_model['input']
572 )
574 @CachedProperty
575 def output_shape(self):
576 if 'output' not in self._operation_model:
577 # Some operations do not define an output shape,
578 # in which case we return None to indicate the
579 # operation has no expected output.
580 return None
581 return self._service_model.resolve_shape_ref(
582 self._operation_model['output']
583 )
585 @CachedProperty
586 def idempotent_members(self):
587 input_shape = self.input_shape
588 if not input_shape:
589 return []
591 return [
592 name
593 for (name, shape) in input_shape.members.items()
594 if 'idempotencyToken' in shape.metadata
595 and shape.metadata['idempotencyToken']
596 ]
598 @CachedProperty
599 def static_context_parameters(self):
600 params = self._operation_model.get('staticContextParams', {})
601 return [
602 StaticContextParameter(name=name, value=props.get('value'))
603 for name, props in params.items()
604 ]
606 @CachedProperty
607 def context_parameters(self):
608 if not self.input_shape:
609 return []
611 return [
612 ContextParameter(
613 name=shape.metadata['contextParam']['name'],
614 member_name=name,
615 )
616 for name, shape in self.input_shape.members.items()
617 if 'contextParam' in shape.metadata
618 and 'name' in shape.metadata['contextParam']
619 ]
621 @CachedProperty
622 def request_compression(self):
623 return self._operation_model.get('requestcompression')
625 @CachedProperty
626 def auth_type(self):
627 return self._operation_model.get('authtype')
629 @CachedProperty
630 def error_shapes(self):
631 shapes = self._operation_model.get("errors", [])
632 return list(self._service_model.resolve_shape_ref(s) for s in shapes)
634 @CachedProperty
635 def endpoint(self):
636 return self._operation_model.get('endpoint')
638 @CachedProperty
639 def http_checksum_required(self):
640 return self._operation_model.get('httpChecksumRequired', False)
642 @CachedProperty
643 def http_checksum(self):
644 return self._operation_model.get('httpChecksum', {})
646 @CachedProperty
647 def has_event_stream_input(self):
648 return self.get_event_stream_input() is not None
650 @CachedProperty
651 def has_event_stream_output(self):
652 return self.get_event_stream_output() is not None
654 def get_event_stream_input(self):
655 return self._get_event_stream(self.input_shape)
657 def get_event_stream_output(self):
658 return self._get_event_stream(self.output_shape)
660 def _get_event_stream(self, shape):
661 """Returns the event stream member's shape if any or None otherwise."""
662 if shape is None:
663 return None
664 event_name = shape.event_stream_name
665 if event_name:
666 return shape.members[event_name]
667 return None
669 @CachedProperty
670 def has_streaming_input(self):
671 return self.get_streaming_input() is not None
673 @CachedProperty
674 def has_streaming_output(self):
675 return self.get_streaming_output() is not None
677 def get_streaming_input(self):
678 return self._get_streaming_body(self.input_shape)
680 def get_streaming_output(self):
681 return self._get_streaming_body(self.output_shape)
683 def _get_streaming_body(self, shape):
684 """Returns the streaming member's shape if any; or None otherwise."""
685 if shape is None:
686 return None
687 payload = shape.serialization.get('payload')
688 if payload is not None:
689 payload_shape = shape.members[payload]
690 if payload_shape.type_name == 'blob':
691 return payload_shape
692 return None
694 def __repr__(self):
695 return f'{self.__class__.__name__}(name={self.name})'
698class ShapeResolver:
699 """Resolves shape references."""
701 # Any type not in this mapping will default to the Shape class.
702 SHAPE_CLASSES = {
703 'structure': StructureShape,
704 'list': ListShape,
705 'map': MapShape,
706 'string': StringShape,
707 }
709 def __init__(self, shape_map):
710 self._shape_map = shape_map
711 self._shape_cache = {}
713 def get_shape_by_name(self, shape_name, member_traits=None):
714 try:
715 shape_model = self._shape_map[shape_name]
716 except KeyError:
717 raise NoShapeFoundError(shape_name)
718 try:
719 shape_cls = self.SHAPE_CLASSES.get(shape_model['type'], Shape)
720 except KeyError:
721 raise InvalidShapeError(
722 f"Shape is missing required key 'type': {shape_model}"
723 )
724 if member_traits:
725 shape_model = shape_model.copy()
726 shape_model.update(member_traits)
727 result = shape_cls(shape_name, shape_model, self)
728 return result
730 def resolve_shape_ref(self, shape_ref):
731 # A shape_ref is a dict that has a 'shape' key that
732 # refers to a shape name as well as any additional
733 # member traits that are then merged over the shape
734 # definition. For example:
735 # {"shape": "StringType", "locationName": "Foobar"}
736 if len(shape_ref) == 1 and 'shape' in shape_ref:
737 # It's just a shape ref with no member traits, we can avoid
738 # a .copy(). This is the common case so it's specifically
739 # called out here.
740 return self.get_shape_by_name(shape_ref['shape'])
741 else:
742 member_traits = shape_ref.copy()
743 try:
744 shape_name = member_traits.pop('shape')
745 except KeyError:
746 raise InvalidShapeReferenceError(
747 f"Invalid model, missing shape reference: {shape_ref}"
748 )
749 return self.get_shape_by_name(shape_name, member_traits)
752class UnresolvableShapeMap:
753 """A ShapeResolver that will throw ValueErrors when shapes are resolved."""
755 def get_shape_by_name(self, shape_name, member_traits=None):
756 raise ValueError(
757 f"Attempted to lookup shape '{shape_name}', but no shape map was provided."
758 )
760 def resolve_shape_ref(self, shape_ref):
761 raise ValueError(
762 f"Attempted to resolve shape '{shape_ref}', but no shape "
763 f"map was provided."
764 )
767class DenormalizedStructureBuilder:
768 """Build a StructureShape from a denormalized model.
770 This is a convenience builder class that makes it easy to construct
771 ``StructureShape``s based on a denormalized model.
773 It will handle the details of creating unique shape names and creating
774 the appropriate shape map needed by the ``StructureShape`` class.
776 Example usage::
778 builder = DenormalizedStructureBuilder()
779 shape = builder.with_members({
780 'A': {
781 'type': 'structure',
782 'members': {
783 'B': {
784 'type': 'structure',
785 'members': {
786 'C': {
787 'type': 'string',
788 }
789 }
790 }
791 }
792 }
793 }).build_model()
794 # ``shape`` is now an instance of botocore.model.StructureShape
796 :type dict_type: class
797 :param dict_type: The dictionary type to use, allowing you to opt-in
798 to using OrderedDict or another dict type. This can
799 be particularly useful for testing when order
800 matters, such as for documentation.
802 """
804 SCALAR_TYPES = (
805 'string',
806 'integer',
807 'boolean',
808 'blob',
809 'float',
810 'timestamp',
811 'long',
812 'double',
813 'char',
814 )
816 def __init__(self, name=None):
817 self.members = OrderedDict()
818 self._name_generator = ShapeNameGenerator()
819 if name is None:
820 self.name = self._name_generator.new_shape_name('structure')
822 def with_members(self, members):
823 """
825 :type members: dict
826 :param members: The denormalized members.
828 :return: self
830 """
831 self._members = members
832 return self
834 def build_model(self):
835 """Build the model based on the provided members.
837 :rtype: botocore.model.StructureShape
838 :return: The built StructureShape object.
840 """
841 shapes = OrderedDict()
842 denormalized = {
843 'type': 'structure',
844 'members': self._members,
845 }
846 self._build_model(denormalized, shapes, self.name)
847 resolver = ShapeResolver(shape_map=shapes)
848 return StructureShape(
849 shape_name=self.name,
850 shape_model=shapes[self.name],
851 shape_resolver=resolver,
852 )
854 def _build_model(self, model, shapes, shape_name):
855 if model['type'] == 'structure':
856 shapes[shape_name] = self._build_structure(model, shapes)
857 elif model['type'] == 'list':
858 shapes[shape_name] = self._build_list(model, shapes)
859 elif model['type'] == 'map':
860 shapes[shape_name] = self._build_map(model, shapes)
861 elif model['type'] in self.SCALAR_TYPES:
862 shapes[shape_name] = self._build_scalar(model)
863 else:
864 raise InvalidShapeError(f"Unknown shape type: {model['type']}")
866 def _build_structure(self, model, shapes):
867 members = OrderedDict()
868 shape = self._build_initial_shape(model)
869 shape['members'] = members
871 for name, member_model in model.get('members', OrderedDict()).items():
872 member_shape_name = self._get_shape_name(member_model)
873 members[name] = {'shape': member_shape_name}
874 self._build_model(member_model, shapes, member_shape_name)
875 return shape
877 def _build_list(self, model, shapes):
878 member_shape_name = self._get_shape_name(model)
879 shape = self._build_initial_shape(model)
880 shape['member'] = {'shape': member_shape_name}
881 self._build_model(model['member'], shapes, member_shape_name)
882 return shape
884 def _build_map(self, model, shapes):
885 key_shape_name = self._get_shape_name(model['key'])
886 value_shape_name = self._get_shape_name(model['value'])
887 shape = self._build_initial_shape(model)
888 shape['key'] = {'shape': key_shape_name}
889 shape['value'] = {'shape': value_shape_name}
890 self._build_model(model['key'], shapes, key_shape_name)
891 self._build_model(model['value'], shapes, value_shape_name)
892 return shape
894 def _build_initial_shape(self, model):
895 shape = {
896 'type': model['type'],
897 }
898 if 'documentation' in model:
899 shape['documentation'] = model['documentation']
900 for attr in Shape.METADATA_ATTRS:
901 if attr in model:
902 shape[attr] = model[attr]
903 return shape
905 def _build_scalar(self, model):
906 return self._build_initial_shape(model)
908 def _get_shape_name(self, model):
909 if 'shape_name' in model:
910 return model['shape_name']
911 else:
912 return self._name_generator.new_shape_name(model['type'])
915class ShapeNameGenerator:
916 """Generate unique shape names for a type.
918 This class can be used in conjunction with the DenormalizedStructureBuilder
919 to generate unique shape names for a given type.
921 """
923 def __init__(self):
924 self._name_cache = defaultdict(int)
926 def new_shape_name(self, type_name):
927 """Generate a unique shape name.
929 This method will guarantee a unique shape name each time it is
930 called with the same type.
932 ::
934 >>> s = ShapeNameGenerator()
935 >>> s.new_shape_name('structure')
936 'StructureType1'
937 >>> s.new_shape_name('structure')
938 'StructureType2'
939 >>> s.new_shape_name('list')
940 'ListType1'
941 >>> s.new_shape_name('list')
942 'ListType2'
945 :type type_name: string
946 :param type_name: The type name (structure, list, map, string, etc.)
948 :rtype: string
949 :return: A unique shape name for the given type
951 """
952 self._name_cache[type_name] += 1
953 current_index = self._name_cache[type_name]
954 return f'{type_name.capitalize()}Type{current_index}'