Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/botocore/model.py: 53%
431 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:03 +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 ]
91 MAP_TYPE = OrderedDict
93 def __init__(self, shape_name, shape_model, shape_resolver=None):
94 """
96 :type shape_name: string
97 :param shape_name: The name of the shape.
99 :type shape_model: dict
100 :param shape_model: The shape model. This would be the value
101 associated with the key in the "shapes" dict of the
102 service model (i.e ``model['shapes'][shape_name]``)
104 :type shape_resolver: botocore.model.ShapeResolver
105 :param shape_resolver: A shape resolver object. This is used to
106 resolve references to other shapes. For scalar shape types
107 (string, integer, boolean, etc.), this argument is not
108 required. If a shape_resolver is not provided for a complex
109 type, then a ``ValueError`` will be raised when an attempt
110 to resolve a shape is made.
112 """
113 self.name = shape_name
114 self.type_name = shape_model['type']
115 self.documentation = shape_model.get('documentation', '')
116 self._shape_model = shape_model
117 if shape_resolver is None:
118 # If a shape_resolver is not provided, we create an object
119 # that will throw errors if you attempt to resolve
120 # a shape. This is actually ok for scalar shapes
121 # because they don't need to resolve shapes and shouldn't
122 # be required to provide an object they won't use.
123 shape_resolver = UnresolvableShapeMap()
124 self._shape_resolver = shape_resolver
125 self._cache = {}
127 @CachedProperty
128 def serialization(self):
129 """Serialization information about the shape.
131 This contains information that may be needed for input serialization
132 or response parsing. This can include:
134 * name
135 * queryName
136 * flattened
137 * location
138 * payload
139 * streaming
140 * xmlNamespace
141 * resultWrapper
142 * xmlAttribute
143 * jsonvalue
144 * timestampFormat
146 :rtype: dict
147 :return: Serialization information about the shape.
149 """
150 model = self._shape_model
151 serialization = {}
152 for attr in self.SERIALIZED_ATTRS:
153 if attr in self._shape_model:
154 serialization[attr] = model[attr]
155 # For consistency, locationName is renamed to just 'name'.
156 if 'locationName' in serialization:
157 serialization['name'] = serialization.pop('locationName')
158 return serialization
160 @CachedProperty
161 def metadata(self):
162 """Metadata about the shape.
164 This requires optional information about the shape, including:
166 * min
167 * max
168 * pattern
169 * enum
170 * sensitive
171 * required
172 * idempotencyToken
173 * document
174 * union
176 :rtype: dict
177 :return: Metadata about the shape.
179 """
180 model = self._shape_model
181 metadata = {}
182 for attr in self.METADATA_ATTRS:
183 if attr in self._shape_model:
184 metadata[attr] = model[attr]
185 return metadata
187 @CachedProperty
188 def required_members(self):
189 """A list of members that are required.
191 A structure shape can define members that are required.
192 This value will return a list of required members. If there
193 are no required members an empty list is returned.
195 """
196 return self.metadata.get('required', [])
198 def _resolve_shape_ref(self, shape_ref):
199 return self._shape_resolver.resolve_shape_ref(shape_ref)
201 def __repr__(self):
202 return f"<{self.__class__.__name__}({self.name})>"
204 @property
205 def event_stream_name(self):
206 return None
209class StructureShape(Shape):
210 @CachedProperty
211 def members(self):
212 members = self._shape_model.get('members', self.MAP_TYPE())
213 # The members dict looks like:
214 # 'members': {
215 # 'MemberName': {'shape': 'shapeName'},
216 # 'MemberName2': {'shape': 'shapeName'},
217 # }
218 # We return a dict of member name to Shape object.
219 shape_members = self.MAP_TYPE()
220 for name, shape_ref in members.items():
221 shape_members[name] = self._resolve_shape_ref(shape_ref)
222 return shape_members
224 @CachedProperty
225 def event_stream_name(self):
226 for member_name, member in self.members.items():
227 if member.serialization.get('eventstream'):
228 return member_name
229 return None
231 @CachedProperty
232 def error_code(self):
233 if not self.metadata.get('exception', False):
234 return None
235 error_metadata = self.metadata.get("error", {})
236 code = error_metadata.get("code")
237 if code:
238 return code
239 # Use the exception name if there is no explicit code modeled
240 return self.name
242 @CachedProperty
243 def is_document_type(self):
244 return self.metadata.get('document', False)
246 @CachedProperty
247 def is_tagged_union(self):
248 return self.metadata.get('union', False)
251class ListShape(Shape):
252 @CachedProperty
253 def member(self):
254 return self._resolve_shape_ref(self._shape_model['member'])
257class MapShape(Shape):
258 @CachedProperty
259 def key(self):
260 return self._resolve_shape_ref(self._shape_model['key'])
262 @CachedProperty
263 def value(self):
264 return self._resolve_shape_ref(self._shape_model['value'])
267class StringShape(Shape):
268 @CachedProperty
269 def enum(self):
270 return self.metadata.get('enum', [])
273class StaticContextParameter(NamedTuple):
274 name: str
275 value: Union[bool, str]
278class ContextParameter(NamedTuple):
279 name: str
280 member_name: str
283class ClientContextParameter(NamedTuple):
284 name: str
285 type: str
286 documentation: str
289class ServiceModel:
290 """
292 :ivar service_description: The parsed service description dictionary.
294 """
296 def __init__(self, service_description, service_name=None):
297 """
299 :type service_description: dict
300 :param service_description: The service description model. This value
301 is obtained from a botocore.loader.Loader, or from directly loading
302 the file yourself::
304 service_description = json.load(
305 open('/path/to/service-description-model.json'))
306 model = ServiceModel(service_description)
308 :type service_name: str
309 :param service_name: The name of the service. Normally this is
310 the endpoint prefix defined in the service_description. However,
311 you can override this value to provide a more convenient name.
312 This is done in a few places in botocore (ses instead of email,
313 emr instead of elasticmapreduce). If this value is not provided,
314 it will default to the endpointPrefix defined in the model.
316 """
317 self._service_description = service_description
318 # We want clients to be able to access metadata directly.
319 self.metadata = service_description.get('metadata', {})
320 self._shape_resolver = ShapeResolver(
321 service_description.get('shapes', {})
322 )
323 self._signature_version = NOT_SET
324 self._service_name = service_name
325 self._instance_cache = {}
327 def shape_for(self, shape_name, member_traits=None):
328 return self._shape_resolver.get_shape_by_name(
329 shape_name, member_traits
330 )
332 def shape_for_error_code(self, error_code):
333 return self._error_code_cache.get(error_code, None)
335 @CachedProperty
336 def _error_code_cache(self):
337 error_code_cache = {}
338 for error_shape in self.error_shapes:
339 code = error_shape.error_code
340 error_code_cache[code] = error_shape
341 return error_code_cache
343 def resolve_shape_ref(self, shape_ref):
344 return self._shape_resolver.resolve_shape_ref(shape_ref)
346 @CachedProperty
347 def shape_names(self):
348 return list(self._service_description.get('shapes', {}))
350 @CachedProperty
351 def error_shapes(self):
352 error_shapes = []
353 for shape_name in self.shape_names:
354 error_shape = self.shape_for(shape_name)
355 if error_shape.metadata.get('exception', False):
356 error_shapes.append(error_shape)
357 return error_shapes
359 @instance_cache
360 def operation_model(self, operation_name):
361 try:
362 model = self._service_description['operations'][operation_name]
363 except KeyError:
364 raise OperationNotFoundError(operation_name)
365 return OperationModel(model, self, operation_name)
367 @CachedProperty
368 def documentation(self):
369 return self._service_description.get('documentation', '')
371 @CachedProperty
372 def operation_names(self):
373 return list(self._service_description.get('operations', []))
375 @CachedProperty
376 def service_name(self):
377 """The name of the service.
379 This defaults to the endpointPrefix defined in the service model.
380 However, this value can be overriden when a ``ServiceModel`` is
381 created. If a service_name was not provided when the ``ServiceModel``
382 was created and if there is no endpointPrefix defined in the
383 service model, then an ``UndefinedModelAttributeError`` exception
384 will be raised.
386 """
387 if self._service_name is not None:
388 return self._service_name
389 else:
390 return self.endpoint_prefix
392 @CachedProperty
393 def service_id(self):
394 try:
395 return ServiceId(self._get_metadata_property('serviceId'))
396 except UndefinedModelAttributeError:
397 raise MissingServiceIdError(service_name=self._service_name)
399 @CachedProperty
400 def signing_name(self):
401 """The name to use when computing signatures.
403 If the model does not define a signing name, this
404 value will be the endpoint prefix defined in the model.
405 """
406 signing_name = self.metadata.get('signingName')
407 if signing_name is None:
408 signing_name = self.endpoint_prefix
409 return signing_name
411 @CachedProperty
412 def api_version(self):
413 return self._get_metadata_property('apiVersion')
415 @CachedProperty
416 def protocol(self):
417 return self._get_metadata_property('protocol')
419 @CachedProperty
420 def endpoint_prefix(self):
421 return self._get_metadata_property('endpointPrefix')
423 @CachedProperty
424 def endpoint_discovery_operation(self):
425 for operation in self.operation_names:
426 model = self.operation_model(operation)
427 if model.is_endpoint_discovery_operation:
428 return model
430 @CachedProperty
431 def endpoint_discovery_required(self):
432 for operation in self.operation_names:
433 model = self.operation_model(operation)
434 if (
435 model.endpoint_discovery is not None
436 and model.endpoint_discovery.get('required')
437 ):
438 return True
439 return False
441 @CachedProperty
442 def client_context_parameters(self):
443 params = self._service_description.get('clientContextParams', {})
444 return [
445 ClientContextParameter(
446 name=param_name,
447 type=param_val['type'],
448 documentation=param_val['documentation'],
449 )
450 for param_name, param_val in params.items()
451 ]
453 def _get_metadata_property(self, name):
454 try:
455 return self.metadata[name]
456 except KeyError:
457 raise UndefinedModelAttributeError(
458 f'"{name}" not defined in the metadata of the model: {self}'
459 )
461 # Signature version is one of the rare properties
462 # that can be modified so a CachedProperty is not used here.
464 @property
465 def signature_version(self):
466 if self._signature_version is NOT_SET:
467 signature_version = self.metadata.get('signatureVersion')
468 self._signature_version = signature_version
469 return self._signature_version
471 @signature_version.setter
472 def signature_version(self, value):
473 self._signature_version = value
475 def __repr__(self):
476 return f'{self.__class__.__name__}({self.service_name})'
479class OperationModel:
480 def __init__(self, operation_model, service_model, name=None):
481 """
483 :type operation_model: dict
484 :param operation_model: The operation model. This comes from the
485 service model, and is the value associated with the operation
486 name in the service model (i.e ``model['operations'][op_name]``).
488 :type service_model: botocore.model.ServiceModel
489 :param service_model: The service model associated with the operation.
491 :type name: string
492 :param name: The operation name. This is the operation name exposed to
493 the users of this model. This can potentially be different from
494 the "wire_name", which is the operation name that *must* by
495 provided over the wire. For example, given::
497 "CreateCloudFrontOriginAccessIdentity":{
498 "name":"CreateCloudFrontOriginAccessIdentity2014_11_06",
499 ...
500 }
502 The ``name`` would be ``CreateCloudFrontOriginAccessIdentity``,
503 but the ``self.wire_name`` would be
504 ``CreateCloudFrontOriginAccessIdentity2014_11_06``, which is the
505 value we must send in the corresponding HTTP request.
507 """
508 self._operation_model = operation_model
509 self._service_model = service_model
510 self._api_name = name
511 # Clients can access '.name' to get the operation name
512 # and '.metadata' to get the top level metdata of the service.
513 self._wire_name = operation_model.get('name')
514 self.metadata = service_model.metadata
515 self.http = operation_model.get('http', {})
517 @CachedProperty
518 def name(self):
519 if self._api_name is not None:
520 return self._api_name
521 else:
522 return self.wire_name
524 @property
525 def wire_name(self):
526 """The wire name of the operation.
528 In many situations this is the same value as the
529 ``name``, value, but in some services, the operation name
530 exposed to the user is different from the operaiton name
531 we send across the wire (e.g cloudfront).
533 Any serialization code should use ``wire_name``.
535 """
536 return self._operation_model.get('name')
538 @property
539 def service_model(self):
540 return self._service_model
542 @CachedProperty
543 def documentation(self):
544 return self._operation_model.get('documentation', '')
546 @CachedProperty
547 def deprecated(self):
548 return self._operation_model.get('deprecated', False)
550 @CachedProperty
551 def endpoint_discovery(self):
552 # Explicit None default. An empty dictionary for this trait means it is
553 # enabled but not required to be used.
554 return self._operation_model.get('endpointdiscovery', None)
556 @CachedProperty
557 def is_endpoint_discovery_operation(self):
558 return self._operation_model.get('endpointoperation', False)
560 @CachedProperty
561 def input_shape(self):
562 if 'input' not in self._operation_model:
563 # Some operations do not accept any input and do not define an
564 # input shape.
565 return None
566 return self._service_model.resolve_shape_ref(
567 self._operation_model['input']
568 )
570 @CachedProperty
571 def output_shape(self):
572 if 'output' not in self._operation_model:
573 # Some operations do not define an output shape,
574 # in which case we return None to indicate the
575 # operation has no expected output.
576 return None
577 return self._service_model.resolve_shape_ref(
578 self._operation_model['output']
579 )
581 @CachedProperty
582 def idempotent_members(self):
583 input_shape = self.input_shape
584 if not input_shape:
585 return []
587 return [
588 name
589 for (name, shape) in input_shape.members.items()
590 if 'idempotencyToken' in shape.metadata
591 and shape.metadata['idempotencyToken']
592 ]
594 @CachedProperty
595 def static_context_parameters(self):
596 params = self._operation_model.get('staticContextParams', {})
597 return [
598 StaticContextParameter(name=name, value=props.get('value'))
599 for name, props in params.items()
600 ]
602 @CachedProperty
603 def context_parameters(self):
604 if not self.input_shape:
605 return []
607 return [
608 ContextParameter(
609 name=shape.metadata['contextParam']['name'],
610 member_name=name,
611 )
612 for name, shape in self.input_shape.members.items()
613 if 'contextParam' in shape.metadata
614 and 'name' in shape.metadata['contextParam']
615 ]
617 @CachedProperty
618 def auth_type(self):
619 return self._operation_model.get('authtype')
621 @CachedProperty
622 def error_shapes(self):
623 shapes = self._operation_model.get("errors", [])
624 return list(self._service_model.resolve_shape_ref(s) for s in shapes)
626 @CachedProperty
627 def endpoint(self):
628 return self._operation_model.get('endpoint')
630 @CachedProperty
631 def http_checksum_required(self):
632 return self._operation_model.get('httpChecksumRequired', False)
634 @CachedProperty
635 def http_checksum(self):
636 return self._operation_model.get('httpChecksum', {})
638 @CachedProperty
639 def has_event_stream_input(self):
640 return self.get_event_stream_input() is not None
642 @CachedProperty
643 def has_event_stream_output(self):
644 return self.get_event_stream_output() is not None
646 def get_event_stream_input(self):
647 return self._get_event_stream(self.input_shape)
649 def get_event_stream_output(self):
650 return self._get_event_stream(self.output_shape)
652 def _get_event_stream(self, shape):
653 """Returns the event stream member's shape if any or None otherwise."""
654 if shape is None:
655 return None
656 event_name = shape.event_stream_name
657 if event_name:
658 return shape.members[event_name]
659 return None
661 @CachedProperty
662 def has_streaming_input(self):
663 return self.get_streaming_input() is not None
665 @CachedProperty
666 def has_streaming_output(self):
667 return self.get_streaming_output() is not None
669 def get_streaming_input(self):
670 return self._get_streaming_body(self.input_shape)
672 def get_streaming_output(self):
673 return self._get_streaming_body(self.output_shape)
675 def _get_streaming_body(self, shape):
676 """Returns the streaming member's shape if any; or None otherwise."""
677 if shape is None:
678 return None
679 payload = shape.serialization.get('payload')
680 if payload is not None:
681 payload_shape = shape.members[payload]
682 if payload_shape.type_name == 'blob':
683 return payload_shape
684 return None
686 def __repr__(self):
687 return f'{self.__class__.__name__}(name={self.name})'
690class ShapeResolver:
691 """Resolves shape references."""
693 # Any type not in this mapping will default to the Shape class.
694 SHAPE_CLASSES = {
695 'structure': StructureShape,
696 'list': ListShape,
697 'map': MapShape,
698 'string': StringShape,
699 }
701 def __init__(self, shape_map):
702 self._shape_map = shape_map
703 self._shape_cache = {}
705 def get_shape_by_name(self, shape_name, member_traits=None):
706 try:
707 shape_model = self._shape_map[shape_name]
708 except KeyError:
709 raise NoShapeFoundError(shape_name)
710 try:
711 shape_cls = self.SHAPE_CLASSES.get(shape_model['type'], Shape)
712 except KeyError:
713 raise InvalidShapeError(
714 f"Shape is missing required key 'type': {shape_model}"
715 )
716 if member_traits:
717 shape_model = shape_model.copy()
718 shape_model.update(member_traits)
719 result = shape_cls(shape_name, shape_model, self)
720 return result
722 def resolve_shape_ref(self, shape_ref):
723 # A shape_ref is a dict that has a 'shape' key that
724 # refers to a shape name as well as any additional
725 # member traits that are then merged over the shape
726 # definition. For example:
727 # {"shape": "StringType", "locationName": "Foobar"}
728 if len(shape_ref) == 1 and 'shape' in shape_ref:
729 # It's just a shape ref with no member traits, we can avoid
730 # a .copy(). This is the common case so it's specifically
731 # called out here.
732 return self.get_shape_by_name(shape_ref['shape'])
733 else:
734 member_traits = shape_ref.copy()
735 try:
736 shape_name = member_traits.pop('shape')
737 except KeyError:
738 raise InvalidShapeReferenceError(
739 f"Invalid model, missing shape reference: {shape_ref}"
740 )
741 return self.get_shape_by_name(shape_name, member_traits)
744class UnresolvableShapeMap:
745 """A ShapeResolver that will throw ValueErrors when shapes are resolved."""
747 def get_shape_by_name(self, shape_name, member_traits=None):
748 raise ValueError(
749 f"Attempted to lookup shape '{shape_name}', but no shape map was provided."
750 )
752 def resolve_shape_ref(self, shape_ref):
753 raise ValueError(
754 f"Attempted to resolve shape '{shape_ref}', but no shape "
755 f"map was provided."
756 )
759class DenormalizedStructureBuilder:
760 """Build a StructureShape from a denormalized model.
762 This is a convenience builder class that makes it easy to construct
763 ``StructureShape``s based on a denormalized model.
765 It will handle the details of creating unique shape names and creating
766 the appropriate shape map needed by the ``StructureShape`` class.
768 Example usage::
770 builder = DenormalizedStructureBuilder()
771 shape = builder.with_members({
772 'A': {
773 'type': 'structure',
774 'members': {
775 'B': {
776 'type': 'structure',
777 'members': {
778 'C': {
779 'type': 'string',
780 }
781 }
782 }
783 }
784 }
785 }).build_model()
786 # ``shape`` is now an instance of botocore.model.StructureShape
788 :type dict_type: class
789 :param dict_type: The dictionary type to use, allowing you to opt-in
790 to using OrderedDict or another dict type. This can
791 be particularly useful for testing when order
792 matters, such as for documentation.
794 """
796 SCALAR_TYPES = (
797 'string',
798 'integer',
799 'boolean',
800 'blob',
801 'float',
802 'timestamp',
803 'long',
804 'double',
805 'char',
806 )
808 def __init__(self, name=None):
809 self.members = OrderedDict()
810 self._name_generator = ShapeNameGenerator()
811 if name is None:
812 self.name = self._name_generator.new_shape_name('structure')
814 def with_members(self, members):
815 """
817 :type members: dict
818 :param members: The denormalized members.
820 :return: self
822 """
823 self._members = members
824 return self
826 def build_model(self):
827 """Build the model based on the provided members.
829 :rtype: botocore.model.StructureShape
830 :return: The built StructureShape object.
832 """
833 shapes = OrderedDict()
834 denormalized = {
835 'type': 'structure',
836 'members': self._members,
837 }
838 self._build_model(denormalized, shapes, self.name)
839 resolver = ShapeResolver(shape_map=shapes)
840 return StructureShape(
841 shape_name=self.name,
842 shape_model=shapes[self.name],
843 shape_resolver=resolver,
844 )
846 def _build_model(self, model, shapes, shape_name):
847 if model['type'] == 'structure':
848 shapes[shape_name] = self._build_structure(model, shapes)
849 elif model['type'] == 'list':
850 shapes[shape_name] = self._build_list(model, shapes)
851 elif model['type'] == 'map':
852 shapes[shape_name] = self._build_map(model, shapes)
853 elif model['type'] in self.SCALAR_TYPES:
854 shapes[shape_name] = self._build_scalar(model)
855 else:
856 raise InvalidShapeError(f"Unknown shape type: {model['type']}")
858 def _build_structure(self, model, shapes):
859 members = OrderedDict()
860 shape = self._build_initial_shape(model)
861 shape['members'] = members
863 for name, member_model in model.get('members', OrderedDict()).items():
864 member_shape_name = self._get_shape_name(member_model)
865 members[name] = {'shape': member_shape_name}
866 self._build_model(member_model, shapes, member_shape_name)
867 return shape
869 def _build_list(self, model, shapes):
870 member_shape_name = self._get_shape_name(model)
871 shape = self._build_initial_shape(model)
872 shape['member'] = {'shape': member_shape_name}
873 self._build_model(model['member'], shapes, member_shape_name)
874 return shape
876 def _build_map(self, model, shapes):
877 key_shape_name = self._get_shape_name(model['key'])
878 value_shape_name = self._get_shape_name(model['value'])
879 shape = self._build_initial_shape(model)
880 shape['key'] = {'shape': key_shape_name}
881 shape['value'] = {'shape': value_shape_name}
882 self._build_model(model['key'], shapes, key_shape_name)
883 self._build_model(model['value'], shapes, value_shape_name)
884 return shape
886 def _build_initial_shape(self, model):
887 shape = {
888 'type': model['type'],
889 }
890 if 'documentation' in model:
891 shape['documentation'] = model['documentation']
892 for attr in Shape.METADATA_ATTRS:
893 if attr in model:
894 shape[attr] = model[attr]
895 return shape
897 def _build_scalar(self, model):
898 return self._build_initial_shape(model)
900 def _get_shape_name(self, model):
901 if 'shape_name' in model:
902 return model['shape_name']
903 else:
904 return self._name_generator.new_shape_name(model['type'])
907class ShapeNameGenerator:
908 """Generate unique shape names for a type.
910 This class can be used in conjunction with the DenormalizedStructureBuilder
911 to generate unique shape names for a given type.
913 """
915 def __init__(self):
916 self._name_cache = defaultdict(int)
918 def new_shape_name(self, type_name):
919 """Generate a unique shape name.
921 This method will guarantee a unique shape name each time it is
922 called with the same type.
924 ::
926 >>> s = ShapeNameGenerator()
927 >>> s.new_shape_name('structure')
928 'StructureType1'
929 >>> s.new_shape_name('structure')
930 'StructureType2'
931 >>> s.new_shape_name('list')
932 'ListType1'
933 >>> s.new_shape_name('list')
934 'ListType2'
937 :type type_name: string
938 :param type_name: The type name (structure, list, map, string, etc.)
940 :rtype: string
941 :return: A unique shape name for the given type
943 """
944 self._name_cache[type_name] += 1
945 current_index = self._name_cache[type_name]
946 return f'{type_name.capitalize()}Type{current_index}'