Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/botocore/model.py: 53%

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.""" 

14 

15from collections import defaultdict 

16from typing import NamedTuple, Union 

17 

18from botocore.auth import resolve_auth_type 

19from botocore.compat import OrderedDict 

20from botocore.exceptions import ( 

21 MissingServiceIdError, 

22 UndefinedModelAttributeError, 

23 UnsupportedServiceProtocolsError, 

24) 

25from botocore.utils import ( 

26 PRIORITY_ORDERED_SUPPORTED_PROTOCOLS, 

27 CachedProperty, 

28 hyphenize_service_id, 

29 instance_cache, 

30) 

31 

32NOT_SET = object() 

33 

34 

35class NoShapeFoundError(Exception): 

36 pass 

37 

38 

39class InvalidShapeError(Exception): 

40 pass 

41 

42 

43class OperationNotFoundError(Exception): 

44 pass 

45 

46 

47class InvalidShapeReferenceError(Exception): 

48 pass 

49 

50 

51class ServiceId(str): 

52 def hyphenize(self): 

53 return hyphenize_service_id(self) 

54 

55 

56class Shape: 

57 """Object representing a shape from the service model.""" 

58 

59 # To simplify serialization logic, all shape params that are 

60 # related to serialization are moved from the top level hash into 

61 # a 'serialization' hash. This list below contains the names of all 

62 # the attributes that should be moved. 

63 SERIALIZED_ATTRS = [ 

64 'locationName', 

65 'queryName', 

66 'flattened', 

67 'location', 

68 'payload', 

69 'streaming', 

70 'timestampFormat', 

71 'xmlNamespace', 

72 'resultWrapper', 

73 'xmlAttribute', 

74 'eventstream', 

75 'event', 

76 'eventheader', 

77 'eventpayload', 

78 'jsonvalue', 

79 'timestampFormat', 

80 'hostLabel', 

81 ] 

82 METADATA_ATTRS = [ 

83 'required', 

84 'min', 

85 'max', 

86 'pattern', 

87 'sensitive', 

88 'enum', 

89 'idempotencyToken', 

90 'error', 

91 'exception', 

92 'endpointdiscoveryid', 

93 'retryable', 

94 'document', 

95 'union', 

96 'contextParam', 

97 'clientContextParams', 

98 'requiresLength', 

99 ] 

100 MAP_TYPE = OrderedDict 

101 

102 def __init__(self, shape_name, shape_model, shape_resolver=None): 

103 """ 

104 

105 :type shape_name: string 

106 :param shape_name: The name of the shape. 

107 

108 :type shape_model: dict 

109 :param shape_model: The shape model. This would be the value 

110 associated with the key in the "shapes" dict of the 

111 service model (i.e ``model['shapes'][shape_name]``) 

112 

113 :type shape_resolver: botocore.model.ShapeResolver 

114 :param shape_resolver: A shape resolver object. This is used to 

115 resolve references to other shapes. For scalar shape types 

116 (string, integer, boolean, etc.), this argument is not 

117 required. If a shape_resolver is not provided for a complex 

118 type, then a ``ValueError`` will be raised when an attempt 

119 to resolve a shape is made. 

120 

121 """ 

122 self.name = shape_name 

123 self.type_name = shape_model['type'] 

124 self.documentation = shape_model.get('documentation', '') 

125 self._shape_model = shape_model 

126 if shape_resolver is None: 

127 # If a shape_resolver is not provided, we create an object 

128 # that will throw errors if you attempt to resolve 

129 # a shape. This is actually ok for scalar shapes 

130 # because they don't need to resolve shapes and shouldn't 

131 # be required to provide an object they won't use. 

132 shape_resolver = UnresolvableShapeMap() 

133 self._shape_resolver = shape_resolver 

134 self._cache = {} 

135 

136 @CachedProperty 

137 def serialization(self): 

138 """Serialization information about the shape. 

139 

140 This contains information that may be needed for input serialization 

141 or response parsing. This can include: 

142 

143 * name 

144 * queryName 

145 * flattened 

146 * location 

147 * payload 

148 * streaming 

149 * xmlNamespace 

150 * resultWrapper 

151 * xmlAttribute 

152 * jsonvalue 

153 * timestampFormat 

154 

155 :rtype: dict 

156 :return: Serialization information about the shape. 

157 

158 """ 

159 model = self._shape_model 

160 serialization = {} 

161 for attr in self.SERIALIZED_ATTRS: 

162 if attr in self._shape_model: 

163 serialization[attr] = model[attr] 

164 # For consistency, locationName is renamed to just 'name'. 

165 if 'locationName' in serialization: 

166 serialization['name'] = serialization.pop('locationName') 

167 return serialization 

168 

169 @CachedProperty 

170 def metadata(self): 

171 """Metadata about the shape. 

172 

173 This requires optional information about the shape, including: 

174 

175 * min 

176 * max 

177 * pattern 

178 * enum 

179 * sensitive 

180 * required 

181 * idempotencyToken 

182 * document 

183 * union 

184 * contextParam 

185 * clientContextParams 

186 * requiresLength 

187 

188 :rtype: dict 

189 :return: Metadata about the shape. 

190 

191 """ 

192 model = self._shape_model 

193 metadata = {} 

194 for attr in self.METADATA_ATTRS: 

195 if attr in self._shape_model: 

196 metadata[attr] = model[attr] 

197 return metadata 

198 

199 @CachedProperty 

200 def required_members(self): 

201 """A list of members that are required. 

202 

203 A structure shape can define members that are required. 

204 This value will return a list of required members. If there 

205 are no required members an empty list is returned. 

206 

207 """ 

208 return self.metadata.get('required', []) 

209 

210 def _resolve_shape_ref(self, shape_ref): 

211 return self._shape_resolver.resolve_shape_ref(shape_ref) 

212 

213 def __repr__(self): 

214 return f"<{self.__class__.__name__}({self.name})>" 

215 

216 @property 

217 def event_stream_name(self): 

218 return None 

219 

220 

221class StructureShape(Shape): 

222 @CachedProperty 

223 def members(self): 

224 members = self._shape_model.get('members', self.MAP_TYPE()) 

225 # The members dict looks like: 

226 # 'members': { 

227 # 'MemberName': {'shape': 'shapeName'}, 

228 # 'MemberName2': {'shape': 'shapeName'}, 

229 # } 

230 # We return a dict of member name to Shape object. 

231 shape_members = self.MAP_TYPE() 

232 for name, shape_ref in members.items(): 

233 shape_members[name] = self._resolve_shape_ref(shape_ref) 

234 return shape_members 

235 

236 @CachedProperty 

237 def event_stream_name(self): 

238 for member_name, member in self.members.items(): 

239 if member.serialization.get('eventstream'): 

240 return member_name 

241 return None 

242 

243 @CachedProperty 

244 def error_code(self): 

245 if not self.metadata.get('exception', False): 

246 return None 

247 error_metadata = self.metadata.get("error", {}) 

248 code = error_metadata.get("code") 

249 if code: 

250 return code 

251 # Use the exception name if there is no explicit code modeled 

252 return self.name 

253 

254 @CachedProperty 

255 def is_document_type(self): 

256 return self.metadata.get('document', False) 

257 

258 @CachedProperty 

259 def is_tagged_union(self): 

260 return self.metadata.get('union', False) 

261 

262 

263class ListShape(Shape): 

264 @CachedProperty 

265 def member(self): 

266 return self._resolve_shape_ref(self._shape_model['member']) 

267 

268 

269class MapShape(Shape): 

270 @CachedProperty 

271 def key(self): 

272 return self._resolve_shape_ref(self._shape_model['key']) 

273 

274 @CachedProperty 

275 def value(self): 

276 return self._resolve_shape_ref(self._shape_model['value']) 

277 

278 

279class StringShape(Shape): 

280 @CachedProperty 

281 def enum(self): 

282 return self.metadata.get('enum', []) 

283 

284 

285class StaticContextParameter(NamedTuple): 

286 name: str 

287 value: Union[bool, str] 

288 

289 

290class ContextParameter(NamedTuple): 

291 name: str 

292 member_name: str 

293 

294 

295class ClientContextParameter(NamedTuple): 

296 name: str 

297 type: str 

298 documentation: str 

299 

300 

301class ServiceModel: 

302 """ 

303 

304 :ivar service_description: The parsed service description dictionary. 

305 

306 """ 

307 

308 def __init__(self, service_description, service_name=None): 

309 """ 

310 

311 :type service_description: dict 

312 :param service_description: The service description model. This value 

313 is obtained from a botocore.loader.Loader, or from directly loading 

314 the file yourself:: 

315 

316 service_description = json.load( 

317 open('/path/to/service-description-model.json')) 

318 model = ServiceModel(service_description) 

319 

320 :type service_name: str 

321 :param service_name: The name of the service. Normally this is 

322 the endpoint prefix defined in the service_description. However, 

323 you can override this value to provide a more convenient name. 

324 This is done in a few places in botocore (ses instead of email, 

325 emr instead of elasticmapreduce). If this value is not provided, 

326 it will default to the endpointPrefix defined in the model. 

327 

328 """ 

329 self._service_description = service_description 

330 # We want clients to be able to access metadata directly. 

331 self.metadata = service_description.get('metadata', {}) 

332 self._shape_resolver = ShapeResolver( 

333 service_description.get('shapes', {}) 

334 ) 

335 self._signature_version = NOT_SET 

336 self._service_name = service_name 

337 self._instance_cache = {} 

338 

339 def shape_for(self, shape_name, member_traits=None): 

340 return self._shape_resolver.get_shape_by_name( 

341 shape_name, member_traits 

342 ) 

343 

344 def shape_for_error_code(self, error_code): 

345 return self._error_code_cache.get(error_code, None) 

346 

347 @CachedProperty 

348 def _error_code_cache(self): 

349 error_code_cache = {} 

350 for error_shape in self.error_shapes: 

351 code = error_shape.error_code 

352 error_code_cache[code] = error_shape 

353 return error_code_cache 

354 

355 def resolve_shape_ref(self, shape_ref): 

356 return self._shape_resolver.resolve_shape_ref(shape_ref) 

357 

358 @CachedProperty 

359 def shape_names(self): 

360 return list(self._service_description.get('shapes', {})) 

361 

362 @CachedProperty 

363 def error_shapes(self): 

364 error_shapes = [] 

365 for shape_name in self.shape_names: 

366 error_shape = self.shape_for(shape_name) 

367 if error_shape.metadata.get('exception', False): 

368 error_shapes.append(error_shape) 

369 return error_shapes 

370 

371 @instance_cache 

372 def operation_model(self, operation_name): 

373 try: 

374 model = self._service_description['operations'][operation_name] 

375 except KeyError: 

376 raise OperationNotFoundError(operation_name) 

377 return OperationModel(model, self, operation_name) 

378 

379 @CachedProperty 

380 def documentation(self): 

381 return self._service_description.get('documentation', '') 

382 

383 @CachedProperty 

384 def operation_names(self): 

385 return list(self._service_description.get('operations', [])) 

386 

387 @CachedProperty 

388 def service_name(self): 

389 """The name of the service. 

390 

391 This defaults to the endpointPrefix defined in the service model. 

392 However, this value can be overriden when a ``ServiceModel`` is 

393 created. If a service_name was not provided when the ``ServiceModel`` 

394 was created and if there is no endpointPrefix defined in the 

395 service model, then an ``UndefinedModelAttributeError`` exception 

396 will be raised. 

397 

398 """ 

399 if self._service_name is not None: 

400 return self._service_name 

401 else: 

402 return self.endpoint_prefix 

403 

404 @CachedProperty 

405 def service_id(self): 

406 try: 

407 return ServiceId(self._get_metadata_property('serviceId')) 

408 except UndefinedModelAttributeError: 

409 raise MissingServiceIdError(service_name=self._service_name) 

410 

411 @CachedProperty 

412 def signing_name(self): 

413 """The name to use when computing signatures. 

414 

415 If the model does not define a signing name, this 

416 value will be the endpoint prefix defined in the model. 

417 """ 

418 signing_name = self.metadata.get('signingName') 

419 if signing_name is None: 

420 signing_name = self.endpoint_prefix 

421 return signing_name 

422 

423 @CachedProperty 

424 def api_version(self): 

425 return self._get_metadata_property('apiVersion') 

426 

427 @CachedProperty 

428 def protocol(self): 

429 return self._get_metadata_property('protocol') 

430 

431 @CachedProperty 

432 def protocols(self): 

433 return self._get_metadata_property('protocols') 

434 

435 @CachedProperty 

436 def resolved_protocol(self): 

437 # We need to ensure `protocols` exists in the metadata before attempting to 

438 # access it directly since referencing service_model.protocols directly will 

439 # raise an UndefinedModelAttributeError if protocols is not defined 

440 if self.metadata.get('protocols'): 

441 for protocol in PRIORITY_ORDERED_SUPPORTED_PROTOCOLS: 

442 if protocol in self.protocols: 

443 return protocol 

444 raise UnsupportedServiceProtocolsError( 

445 botocore_supported_protocols=PRIORITY_ORDERED_SUPPORTED_PROTOCOLS, 

446 service_supported_protocols=self.protocols, 

447 service=self.service_name, 

448 ) 

449 # If a service does not have a `protocols` trait, fall back to the legacy 

450 # `protocol` trait 

451 return self.protocol 

452 

453 @CachedProperty 

454 def endpoint_prefix(self): 

455 return self._get_metadata_property('endpointPrefix') 

456 

457 @CachedProperty 

458 def endpoint_discovery_operation(self): 

459 for operation in self.operation_names: 

460 model = self.operation_model(operation) 

461 if model.is_endpoint_discovery_operation: 

462 return model 

463 

464 @CachedProperty 

465 def endpoint_discovery_required(self): 

466 for operation in self.operation_names: 

467 model = self.operation_model(operation) 

468 if ( 

469 model.endpoint_discovery is not None 

470 and model.endpoint_discovery.get('required') 

471 ): 

472 return True 

473 return False 

474 

475 @CachedProperty 

476 def client_context_parameters(self): 

477 params = self._service_description.get('clientContextParams', {}) 

478 return [ 

479 ClientContextParameter( 

480 name=param_name, 

481 type=param_val['type'], 

482 documentation=param_val['documentation'], 

483 ) 

484 for param_name, param_val in params.items() 

485 ] 

486 

487 def _get_metadata_property(self, name): 

488 try: 

489 return self.metadata[name] 

490 except KeyError: 

491 raise UndefinedModelAttributeError( 

492 f'"{name}" not defined in the metadata of the model: {self}' 

493 ) 

494 

495 # Signature version is one of the rare properties 

496 # that can be modified so a CachedProperty is not used here. 

497 

498 @property 

499 def signature_version(self): 

500 if self._signature_version is NOT_SET: 

501 signature_version = self.metadata.get('signatureVersion') 

502 self._signature_version = signature_version 

503 return self._signature_version 

504 

505 @signature_version.setter 

506 def signature_version(self, value): 

507 self._signature_version = value 

508 

509 @CachedProperty 

510 def is_query_compatible(self): 

511 return 'awsQueryCompatible' in self.metadata 

512 

513 def __repr__(self): 

514 return f'{self.__class__.__name__}({self.service_name})' 

515 

516 

517class OperationModel: 

518 def __init__(self, operation_model, service_model, name=None): 

519 """ 

520 

521 :type operation_model: dict 

522 :param operation_model: The operation model. This comes from the 

523 service model, and is the value associated with the operation 

524 name in the service model (i.e ``model['operations'][op_name]``). 

525 

526 :type service_model: botocore.model.ServiceModel 

527 :param service_model: The service model associated with the operation. 

528 

529 :type name: string 

530 :param name: The operation name. This is the operation name exposed to 

531 the users of this model. This can potentially be different from 

532 the "wire_name", which is the operation name that *must* by 

533 provided over the wire. For example, given:: 

534 

535 "CreateCloudFrontOriginAccessIdentity":{ 

536 "name":"CreateCloudFrontOriginAccessIdentity2014_11_06", 

537 ... 

538 } 

539 

540 The ``name`` would be ``CreateCloudFrontOriginAccessIdentity``, 

541 but the ``self.wire_name`` would be 

542 ``CreateCloudFrontOriginAccessIdentity2014_11_06``, which is the 

543 value we must send in the corresponding HTTP request. 

544 

545 """ 

546 self._operation_model = operation_model 

547 self._service_model = service_model 

548 self._api_name = name 

549 # Clients can access '.name' to get the operation name 

550 # and '.metadata' to get the top level metdata of the service. 

551 self._wire_name = operation_model.get('name') 

552 self.metadata = service_model.metadata 

553 self.http = operation_model.get('http', {}) 

554 

555 @CachedProperty 

556 def name(self): 

557 if self._api_name is not None: 

558 return self._api_name 

559 else: 

560 return self.wire_name 

561 

562 @property 

563 def wire_name(self): 

564 """The wire name of the operation. 

565 

566 In many situations this is the same value as the 

567 ``name``, value, but in some services, the operation name 

568 exposed to the user is different from the operation name 

569 we send across the wire (e.g cloudfront). 

570 

571 Any serialization code should use ``wire_name``. 

572 

573 """ 

574 return self._operation_model.get('name') 

575 

576 @property 

577 def service_model(self): 

578 return self._service_model 

579 

580 @CachedProperty 

581 def documentation(self): 

582 return self._operation_model.get('documentation', '') 

583 

584 @CachedProperty 

585 def deprecated(self): 

586 return self._operation_model.get('deprecated', False) 

587 

588 @CachedProperty 

589 def endpoint_discovery(self): 

590 # Explicit None default. An empty dictionary for this trait means it is 

591 # enabled but not required to be used. 

592 return self._operation_model.get('endpointdiscovery', None) 

593 

594 @CachedProperty 

595 def is_endpoint_discovery_operation(self): 

596 return self._operation_model.get('endpointoperation', False) 

597 

598 @CachedProperty 

599 def input_shape(self): 

600 if 'input' not in self._operation_model: 

601 # Some operations do not accept any input and do not define an 

602 # input shape. 

603 return None 

604 return self._service_model.resolve_shape_ref( 

605 self._operation_model['input'] 

606 ) 

607 

608 @CachedProperty 

609 def output_shape(self): 

610 if 'output' not in self._operation_model: 

611 # Some operations do not define an output shape, 

612 # in which case we return None to indicate the 

613 # operation has no expected output. 

614 return None 

615 return self._service_model.resolve_shape_ref( 

616 self._operation_model['output'] 

617 ) 

618 

619 @CachedProperty 

620 def idempotent_members(self): 

621 input_shape = self.input_shape 

622 if not input_shape: 

623 return [] 

624 

625 return [ 

626 name 

627 for (name, shape) in input_shape.members.items() 

628 if 'idempotencyToken' in shape.metadata 

629 and shape.metadata['idempotencyToken'] 

630 ] 

631 

632 @CachedProperty 

633 def static_context_parameters(self): 

634 params = self._operation_model.get('staticContextParams', {}) 

635 return [ 

636 StaticContextParameter(name=name, value=props.get('value')) 

637 for name, props in params.items() 

638 ] 

639 

640 @CachedProperty 

641 def context_parameters(self): 

642 if not self.input_shape: 

643 return [] 

644 

645 return [ 

646 ContextParameter( 

647 name=shape.metadata['contextParam']['name'], 

648 member_name=name, 

649 ) 

650 for name, shape in self.input_shape.members.items() 

651 if 'contextParam' in shape.metadata 

652 and 'name' in shape.metadata['contextParam'] 

653 ] 

654 

655 @CachedProperty 

656 def operation_context_parameters(self): 

657 return self._operation_model.get('operationContextParams', []) 

658 

659 @CachedProperty 

660 def request_compression(self): 

661 return self._operation_model.get('requestcompression') 

662 

663 @CachedProperty 

664 def auth(self): 

665 return self._operation_model.get('auth') 

666 

667 @CachedProperty 

668 def auth_type(self): 

669 return self._operation_model.get('authtype') 

670 

671 @CachedProperty 

672 def resolved_auth_type(self): 

673 if self.auth: 

674 return resolve_auth_type(self.auth) 

675 return self.auth_type 

676 

677 @CachedProperty 

678 def unsigned_payload(self): 

679 return self._operation_model.get('unsignedPayload') 

680 

681 @CachedProperty 

682 def error_shapes(self): 

683 shapes = self._operation_model.get("errors", []) 

684 return list(self._service_model.resolve_shape_ref(s) for s in shapes) 

685 

686 @CachedProperty 

687 def endpoint(self): 

688 return self._operation_model.get('endpoint') 

689 

690 @CachedProperty 

691 def http_checksum_required(self): 

692 return self._operation_model.get('httpChecksumRequired', False) 

693 

694 @CachedProperty 

695 def http_checksum(self): 

696 return self._operation_model.get('httpChecksum', {}) 

697 

698 @CachedProperty 

699 def has_event_stream_input(self): 

700 return self.get_event_stream_input() is not None 

701 

702 @CachedProperty 

703 def has_event_stream_output(self): 

704 return self.get_event_stream_output() is not None 

705 

706 def get_event_stream_input(self): 

707 return self._get_event_stream(self.input_shape) 

708 

709 def get_event_stream_output(self): 

710 return self._get_event_stream(self.output_shape) 

711 

712 def _get_event_stream(self, shape): 

713 """Returns the event stream member's shape if any or None otherwise.""" 

714 if shape is None: 

715 return None 

716 event_name = shape.event_stream_name 

717 if event_name: 

718 return shape.members[event_name] 

719 return None 

720 

721 @CachedProperty 

722 def has_streaming_input(self): 

723 return self.get_streaming_input() is not None 

724 

725 @CachedProperty 

726 def has_streaming_output(self): 

727 return self.get_streaming_output() is not None 

728 

729 def get_streaming_input(self): 

730 return self._get_streaming_body(self.input_shape) 

731 

732 def get_streaming_output(self): 

733 return self._get_streaming_body(self.output_shape) 

734 

735 def _get_streaming_body(self, shape): 

736 """Returns the streaming member's shape if any; or None otherwise.""" 

737 if shape is None: 

738 return None 

739 payload = shape.serialization.get('payload') 

740 if payload is not None: 

741 payload_shape = shape.members[payload] 

742 if payload_shape.type_name == 'blob': 

743 return payload_shape 

744 return None 

745 

746 def __repr__(self): 

747 return f'{self.__class__.__name__}(name={self.name})' 

748 

749 

750class ShapeResolver: 

751 """Resolves shape references.""" 

752 

753 # Any type not in this mapping will default to the Shape class. 

754 SHAPE_CLASSES = { 

755 'structure': StructureShape, 

756 'list': ListShape, 

757 'map': MapShape, 

758 'string': StringShape, 

759 } 

760 

761 def __init__(self, shape_map): 

762 self._shape_map = shape_map 

763 self._shape_cache = {} 

764 

765 def get_shape_by_name(self, shape_name, member_traits=None): 

766 try: 

767 shape_model = self._shape_map[shape_name] 

768 except KeyError: 

769 raise NoShapeFoundError(shape_name) 

770 try: 

771 shape_cls = self.SHAPE_CLASSES.get(shape_model['type'], Shape) 

772 except KeyError: 

773 raise InvalidShapeError( 

774 f"Shape is missing required key 'type': {shape_model}" 

775 ) 

776 if member_traits: 

777 shape_model = shape_model.copy() 

778 shape_model.update(member_traits) 

779 result = shape_cls(shape_name, shape_model, self) 

780 return result 

781 

782 def resolve_shape_ref(self, shape_ref): 

783 # A shape_ref is a dict that has a 'shape' key that 

784 # refers to a shape name as well as any additional 

785 # member traits that are then merged over the shape 

786 # definition. For example: 

787 # {"shape": "StringType", "locationName": "Foobar"} 

788 if len(shape_ref) == 1 and 'shape' in shape_ref: 

789 # It's just a shape ref with no member traits, we can avoid 

790 # a .copy(). This is the common case so it's specifically 

791 # called out here. 

792 return self.get_shape_by_name(shape_ref['shape']) 

793 else: 

794 member_traits = shape_ref.copy() 

795 try: 

796 shape_name = member_traits.pop('shape') 

797 except KeyError: 

798 raise InvalidShapeReferenceError( 

799 f"Invalid model, missing shape reference: {shape_ref}" 

800 ) 

801 return self.get_shape_by_name(shape_name, member_traits) 

802 

803 

804class UnresolvableShapeMap: 

805 """A ShapeResolver that will throw ValueErrors when shapes are resolved.""" 

806 

807 def get_shape_by_name(self, shape_name, member_traits=None): 

808 raise ValueError( 

809 f"Attempted to lookup shape '{shape_name}', but no shape map was provided." 

810 ) 

811 

812 def resolve_shape_ref(self, shape_ref): 

813 raise ValueError( 

814 f"Attempted to resolve shape '{shape_ref}', but no shape " 

815 f"map was provided." 

816 ) 

817 

818 

819class DenormalizedStructureBuilder: 

820 """Build a StructureShape from a denormalized model. 

821 

822 This is a convenience builder class that makes it easy to construct 

823 ``StructureShape``s based on a denormalized model. 

824 

825 It will handle the details of creating unique shape names and creating 

826 the appropriate shape map needed by the ``StructureShape`` class. 

827 

828 Example usage:: 

829 

830 builder = DenormalizedStructureBuilder() 

831 shape = builder.with_members({ 

832 'A': { 

833 'type': 'structure', 

834 'members': { 

835 'B': { 

836 'type': 'structure', 

837 'members': { 

838 'C': { 

839 'type': 'string', 

840 } 

841 } 

842 } 

843 } 

844 } 

845 }).build_model() 

846 # ``shape`` is now an instance of botocore.model.StructureShape 

847 

848 :type dict_type: class 

849 :param dict_type: The dictionary type to use, allowing you to opt-in 

850 to using OrderedDict or another dict type. This can 

851 be particularly useful for testing when order 

852 matters, such as for documentation. 

853 

854 """ 

855 

856 SCALAR_TYPES = ( 

857 'string', 

858 'integer', 

859 'boolean', 

860 'blob', 

861 'float', 

862 'timestamp', 

863 'long', 

864 'double', 

865 'char', 

866 ) 

867 

868 def __init__(self, name=None): 

869 self.members = OrderedDict() 

870 self._name_generator = ShapeNameGenerator() 

871 if name is None: 

872 self.name = self._name_generator.new_shape_name('structure') 

873 

874 def with_members(self, members): 

875 """ 

876 

877 :type members: dict 

878 :param members: The denormalized members. 

879 

880 :return: self 

881 

882 """ 

883 self._members = members 

884 return self 

885 

886 def build_model(self): 

887 """Build the model based on the provided members. 

888 

889 :rtype: botocore.model.StructureShape 

890 :return: The built StructureShape object. 

891 

892 """ 

893 shapes = OrderedDict() 

894 denormalized = { 

895 'type': 'structure', 

896 'members': self._members, 

897 } 

898 self._build_model(denormalized, shapes, self.name) 

899 resolver = ShapeResolver(shape_map=shapes) 

900 return StructureShape( 

901 shape_name=self.name, 

902 shape_model=shapes[self.name], 

903 shape_resolver=resolver, 

904 ) 

905 

906 def _build_model(self, model, shapes, shape_name): 

907 if model['type'] == 'structure': 

908 shapes[shape_name] = self._build_structure(model, shapes) 

909 elif model['type'] == 'list': 

910 shapes[shape_name] = self._build_list(model, shapes) 

911 elif model['type'] == 'map': 

912 shapes[shape_name] = self._build_map(model, shapes) 

913 elif model['type'] in self.SCALAR_TYPES: 

914 shapes[shape_name] = self._build_scalar(model) 

915 else: 

916 raise InvalidShapeError(f"Unknown shape type: {model['type']}") 

917 

918 def _build_structure(self, model, shapes): 

919 members = OrderedDict() 

920 shape = self._build_initial_shape(model) 

921 shape['members'] = members 

922 

923 for name, member_model in model.get('members', OrderedDict()).items(): 

924 member_shape_name = self._get_shape_name(member_model) 

925 members[name] = {'shape': member_shape_name} 

926 self._build_model(member_model, shapes, member_shape_name) 

927 return shape 

928 

929 def _build_list(self, model, shapes): 

930 member_shape_name = self._get_shape_name(model) 

931 shape = self._build_initial_shape(model) 

932 shape['member'] = {'shape': member_shape_name} 

933 self._build_model(model['member'], shapes, member_shape_name) 

934 return shape 

935 

936 def _build_map(self, model, shapes): 

937 key_shape_name = self._get_shape_name(model['key']) 

938 value_shape_name = self._get_shape_name(model['value']) 

939 shape = self._build_initial_shape(model) 

940 shape['key'] = {'shape': key_shape_name} 

941 shape['value'] = {'shape': value_shape_name} 

942 self._build_model(model['key'], shapes, key_shape_name) 

943 self._build_model(model['value'], shapes, value_shape_name) 

944 return shape 

945 

946 def _build_initial_shape(self, model): 

947 shape = { 

948 'type': model['type'], 

949 } 

950 if 'documentation' in model: