Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py # Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License.	562 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 conditions.py # Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import re from collections import namedtuple from boto3.exceptions import ( DynamoDBNeedsConditionError, DynamoDBNeedsKeyConditionError, DynamoDBOperationNotSupportedError, ) ATTR_NAME_REGEX = re.compile(r'[^.\[\]]+(?![^\[]\])') class ConditionBase: expression_format = '' expression_operator = '' has_grouped_values = False def __init__(self, values): self._values = values def __and__(self, other): if not isinstance(other, ConditionBase): raise DynamoDBOperationNotSupportedError('AND', other) return And(self, other) def __or__(self, other): if not isinstance(other, ConditionBase): raise DynamoDBOperationNotSupportedError('OR', other) return Or(self, other) def __invert__(self): return Not(self) def get_expression(self): return { 'format': self.expression_format, 'operator': self.expression_operator, 'values': self._values, } def __eq__(self, other): if isinstance(other, type(self)): if self._values == other._values: return True return False def __ne__(self, other): return not self.__eq__(other) class AttributeBase: def __init__(self, name): self.name = name def __and__(self, value): raise DynamoDBOperationNotSupportedError('AND', self) def __or__(self, value): raise DynamoDBOperationNotSupportedError('OR', self) def __invert__(self): raise DynamoDBOperationNotSupportedError('NOT', self) def eq(self, value): """Creates a condition where the attribute is equal to the value. :param value: The value that the attribute is equal to. """ return Equals(self, value) def lt(self, value): """Creates a condition where the attribute is less than the value. :param value: The value that the attribute is less than. """ return LessThan(self, value) def lte(self, value): """Creates a condition where the attribute is less than or equal to the value. :param value: The value that the attribute is less than or equal to. """ return LessThanEquals(self, value) def gt(self, value): """Creates a condition where the attribute is greater than the value. :param value: The value that the attribute is greater than. """ return GreaterThan(self, value) def gte(self, value): """Creates a condition where the attribute is greater than or equal to the value. :param value: The value that the attribute is greater than or equal to. """ return GreaterThanEquals(self, value) def begins_with(self, value): """Creates a condition where the attribute begins with the value. :param value: The value that the attribute begins with. """ return BeginsWith(self, value) def between(self, low_value, high_value): """Creates a condition where the attribute is greater than or equal to the low value and less than or equal to the high value. :param low_value: The value that the attribute is greater than or equal to. :param high_value: The value that the attribute is less than or equal to. """ return Between(self, low_value, high_value) def __eq__(self, other): return isinstance(other, type(self)) and self.name == other.name def __ne__(self, other): return not self.__eq__(other) class ConditionAttributeBase(ConditionBase, AttributeBase): """This base class is for conditions that can have attribute methods. One example is the Size condition. To complete a condition, you need to apply another AttributeBase method like eq(). """ def __init__(self, values): ConditionBase.__init__(self, values) # This is assuming the first value to the condition is the attribute # in which can be used to generate its attribute base. AttributeBase.__init__(self, values[0].name) def __eq__(self, other): return ConditionBase.__eq__(self, other) and AttributeBase.__eq__( self, other ) def __ne__(self, other): return not self.__eq__(other) class ComparisonCondition(ConditionBase): expression_format = '{0} {operator} {1}' class Equals(ComparisonCondition): expression_operator = '=' class NotEquals(ComparisonCondition): expression_operator = '<>' class LessThan(ComparisonCondition): expression_operator = '<' class LessThanEquals(ComparisonCondition): expression_operator = '<=' class GreaterThan(ComparisonCondition): expression_operator = '>' class GreaterThanEquals(ComparisonCondition): expression_operator = '>=' class In(ComparisonCondition): expression_operator = 'IN' has_grouped_values = True class Between(ConditionBase): expression_operator = 'BETWEEN' expression_format = '{0} {operator} {1} AND {2}' class BeginsWith(ConditionBase): expression_operator = 'begins_with' expression_format = '{operator}({0}, {1})' class Contains(ConditionBase): expression_operator = 'contains' expression_format = '{operator}({0}, {1})' class Size(ConditionAttributeBase): expression_operator = 'size' expression_format = '{operator}({0})' class AttributeType(ConditionBase): expression_operator = 'attribute_type' expression_format = '{operator}({0}, {1})' class AttributeExists(ConditionBase): expression_operator = 'attribute_exists' expression_format = '{operator}({0})' class AttributeNotExists(ConditionBase): expression_operator = 'attribute_not_exists' expression_format = '{operator}({0})' class And(ConditionBase): expression_operator = 'AND' expression_format = '({0} {operator} {1})' class Or(ConditionBase): expression_operator = 'OR' expression_format = '({0} {operator} {1})' class Not(ConditionBase): expression_operator = 'NOT' expression_format = '({operator} {0})' class Key(AttributeBase): pass class Attr(AttributeBase): """Represents an DynamoDB item's attribute.""" def ne(self, value): """Creates a condition where the attribute is not equal to the value :param value: The value that the attribute is not equal to. """ return NotEquals(self, value) def is_in(self, value): """Creates a condition where the attribute is in the value, :type value: list :param value: The value that the attribute is in. """ return In(self, value) def exists(self): """Creates a condition where the attribute exists.""" return AttributeExists(self) def not_exists(self): """Creates a condition where the attribute does not exist.""" return AttributeNotExists(self) def contains(self, value): """Creates a condition where the attribute contains the value. :param value: The value the attribute contains. """ return Contains(self, value) def size(self): """Creates a condition for the attribute size. Note another AttributeBase method must be called on the returned size condition to be a valid DynamoDB condition. """ return Size(self) def attribute_type(self, value): """Creates a condition for the attribute type. :param value: The type of the attribute. """ return AttributeType(self, value) BuiltConditionExpression = namedtuple( 'BuiltConditionExpression', [ 'condition_expression', 'attribute_name_placeholders', 'attribute_value_placeholders', ], ) class ConditionExpressionBuilder: """This class is used to build condition expressions with placeholders""" def __init__(self): self._name_count = 0 self._value_count = 0 self._name_placeholder = 'n' self._value_placeholder = 'v' def _get_name_placeholder(self): return '#' + self._name_placeholder + str(self._name_count) def _get_value_placeholder(self): return ':' + self._value_placeholder + str(self._value_count) def reset(self): """Resets the placeholder name and values""" self._name_count = 0 self._value_count = 0 def build_expression(self, condition, is_key_condition=False): """Builds the condition expression and the dictionary of placeholders. :type condition: ConditionBase :param condition: A condition to be built into a condition expression string with any necessary placeholders. :type is_key_condition: Boolean :param is_key_condition: True if the expression is for a KeyConditionExpression. False otherwise. :rtype: (string, dict, dict) :returns: Will return a string representing the condition with placeholders inserted where necessary, a dictionary of placeholders for attribute names, and a dictionary of placeholders for attribute values. Here is a sample return value: ('#n0 = :v0', {'#n0': 'myattribute'}, {':v1': 'myvalue'}) """ if not isinstance(condition, ConditionBase): raise DynamoDBNeedsConditionError(condition) attribute_name_placeholders = {} attribute_value_placeholders = {} condition_expression = self._build_expression( condition, attribute_name_placeholders, attribute_value_placeholders, is_key_condition=is_key_condition, ) return BuiltConditionExpression( condition_expression=condition_expression, attribute_name_placeholders=attribute_name_placeholders, attribute_value_placeholders=attribute_value_placeholders, ) def _build_expression( self, condition, attribute_name_placeholders, attribute_value_placeholders, is_key_condition, ): expression_dict = condition.get_expression() replaced_values = [] for value in expression_dict['values']: # Build the necessary placeholders for that value. # Placeholders are built for both attribute names and values. replaced_value = self._build_expression_component( value, attribute_name_placeholders, attribute_value_placeholders, condition.has_grouped_values, is_key_condition, ) replaced_values.append(replaced_value) # Fill out the expression using the operator and the # values that have been replaced with placeholders. return expression_dict['format'].format( *replaced_values, operator=expression_dict['operator'] ) def _build_expression_component( self, value, attribute_name_placeholders, attribute_value_placeholders, has_grouped_values, is_key_condition, ): # Continue to recurse if the value is a ConditionBase in order # to extract out all parts of the expression. if isinstance(value, ConditionBase): return self._build_expression( value, attribute_name_placeholders, attribute_value_placeholders, is_key_condition, ) # If it is not a ConditionBase, we can recurse no further. # So we check if it is an attribute and add placeholders for # its name elif isinstance(value, AttributeBase): if is_key_condition and not isinstance(value, Key): raise DynamoDBNeedsKeyConditionError( f'Attribute object {value.name} is of type {type(value)}. ' f'KeyConditionExpression only supports Attribute objects ' f'of type Key' ) return self._build_name_placeholder( value, attribute_name_placeholders ) # If it is anything else, we treat it as a value and thus placeholders # are needed for the value. else: return self._build_value_placeholder( value, attribute_value_placeholders, has_grouped_values ) def _build_name_placeholder(self, value, attribute_name_placeholders): attribute_name = value.name # Figure out which parts of the attribute name that needs replacement. attribute_name_parts = ATTR_NAME_REGEX.findall(attribute_name) # Add a temporary placeholder for each of these parts. placeholder_format = ATTR_NAME_REGEX.sub('%s', attribute_name) str_format_args = [] for part in attribute_name_parts: name_placeholder = self._get_name_placeholder() self._name_count += 1 str_format_args.append(name_placeholder) # Add the placeholder and value to dictionary of name placeholders. attribute_name_placeholders[name_placeholder] = part # Replace the temporary placeholders with the designated placeholders. return placeholder_format % tuple(str_format_args) def _build_value_placeholder( self, value, attribute_value_placeholders, has_grouped_values=False ): # If the values are grouped, we need to add a placeholder for # each element inside of the actual value. if has_grouped_values: placeholder_list = [] for v in value: value_placeholder = self._get_value_placeholder() self._value_count += 1 placeholder_list.append(value_placeholder) attribute_value_placeholders[value_placeholder] = v # Assuming the values are grouped by parenthesis. # IN is the currently the only one that uses this so it maybe # needed to be changed in future. return '(' + ', '.join(placeholder_list) + ')' # Otherwise, treat the value as a single value that needs only # one placeholder. else: value_placeholder = self._get_value_placeholder() self._value_count += 1 attribute_value_placeholders[value_placeholder] = value return value_placeholder	14.69 KB	2025-05-25 12:26
📄 table.py # Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import logging logger = logging.getLogger(__name__) def register_table_methods(base_classes, *kwargs): base_classes.insert(0, TableResource) # This class can be used to add any additional methods we want # onto a table resource. Ideally to avoid creating a new # base class for every method we can just update this # class instead. Just be sure to move the bulk of the # actual method implementation to another class. class TableResource: def __init__(self, args, *kwargs): super().__init__(args, kwargs) def batch_writer(self, overwrite_by_pkeys=None): """Create a batch writer object. This method creates a context manager for writing objects to Amazon DynamoDB in batch. The batch writer will automatically handle buffering and sending items in batches. In addition, the batch writer will also automatically handle any unprocessed items and resend them as needed. All you need to do is call ``put_item`` for any items you want to add, and ``delete_item`` for any items you want to delete. Example usage:: with table.batch_writer() as batch: for _ in range(1000000): batch.put_item(Item={'HashKey': '...', 'Otherstuff': '...'}) # You can also delete_items in a batch. batch.delete_item(Key={'HashKey': 'SomeHashKey'}) :type overwrite_by_pkeys: list(string) :param overwrite_by_pkeys: De-duplicate request items in buffer if match new request item on specified primary keys. i.e ``["partition_key1", "sort_key2", "sort_key3"]`` """ return BatchWriter( self.name, self.meta.client, overwrite_by_pkeys=overwrite_by_pkeys ) class BatchWriter: """Automatically handle batch writes to DynamoDB for a single table.""" def __init__( self, table_name, client, flush_amount=25, overwrite_by_pkeys=None ): """ :type table_name: str :param table_name: The name of the table. The class handles batch writes to a single table. :type client: ``botocore.client.Client`` :param client: A botocore client. Note this client must** have the dynamodb customizations applied to it for transforming AttributeValues into the wire protocol. What this means in practice is that you need to use a client that comes from a DynamoDB resource if you're going to instantiate this class directly, i.e ``boto3.resource('dynamodb').Table('foo').meta.client``. :type flush_amount: int :param flush_amount: The number of items to keep in a local buffer before sending a batch_write_item request to DynamoDB. :type overwrite_by_pkeys: list(string) :param overwrite_by_pkeys: De-duplicate request items in buffer if match new request item on specified primary keys. i.e ``["partition_key1", "sort_key2", "sort_key3"]`` """ self._table_name = table_name self._client = client self._items_buffer = [] self._flush_amount = flush_amount self._overwrite_by_pkeys = overwrite_by_pkeys def put_item(self, Item): self._add_request_and_process({'PutRequest': {'Item': Item}}) def delete_item(self, Key): self._add_request_and_process({'DeleteRequest': {'Key': Key}}) def _add_request_and_process(self, request): if self._overwrite_by_pkeys: self._remove_dup_pkeys_request_if_any(request) self._items_buffer.append(request) self._flush_if_needed() def _remove_dup_pkeys_request_if_any(self, request): pkey_values_new = self._extract_pkey_values(request) for item in self._items_buffer: if self._extract_pkey_values(item) == pkey_values_new: self._items_buffer.remove(item) logger.debug( "With overwrite_by_pkeys enabled, skipping " "request:%s", item, ) def _extract_pkey_values(self, request): if request.get('PutRequest'): return [ request['PutRequest']['Item'][key] for key in self._overwrite_by_pkeys ] elif request.get('DeleteRequest'): return [ request['DeleteRequest']['Key'][key] for key in self._overwrite_by_pkeys ] return None def _flush_if_needed(self): if len(self._items_buffer) >= self._flush_amount: self._flush() def _flush(self): items_to_send = self._items_buffer[: self._flush_amount] self._items_buffer = self._items_buffer[self._flush_amount :] response = self._client.batch_write_item( RequestItems={self._table_name: items_to_send} ) unprocessed_items = response['UnprocessedItems'] if not unprocessed_items: unprocessed_items = {} item_list = unprocessed_items.get(self._table_name, []) # Any unprocessed_items are immediately added to the # next batch we send. self._items_buffer.extend(item_list) logger.debug( "Batch write sent %s, unprocessed: %s", len(items_to_send), len(self._items_buffer), ) def __enter__(self): return self def __exit__(self, exc_type, exc_value, tb): # When we exit, we need to keep flushing whatever's left # until there's nothing left in our items buffer. while self._items_buffer: self._flush()	6.19 KB	2025-05-25 12:26
📄 transform.py # Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import copy from boto3.compat import collections_abc from boto3.docs.utils import DocumentModifiedShape from boto3.dynamodb.conditions import ConditionBase, ConditionExpressionBuilder from boto3.dynamodb.types import TypeDeserializer, TypeSerializer def register_high_level_interface(base_classes, kwargs): base_classes.insert(0, DynamoDBHighLevelResource) class _ForgetfulDict(dict): """A dictionary that discards any items set on it. For use as `memo` in `copy.deepcopy()` when every instance of a repeated object in the deepcopied data structure should result in a separate copy. """ def __setitem__(self, key, value): pass def copy_dynamodb_params(params, kwargs): return copy.deepcopy(params, memo=_ForgetfulDict()) class DynamoDBHighLevelResource: def __init__(self, args, kwargs): super().__init__(args, *kwargs) # Apply handler that creates a copy of the user provided dynamodb # item such that it can be modified. self.meta.client.meta.events.register( 'provide-client-params.dynamodb', copy_dynamodb_params, unique_id='dynamodb-create-params-copy', ) self._injector = TransformationInjector() # Apply the handler that generates condition expressions including # placeholders. self.meta.client.meta.events.register( 'before-parameter-build.dynamodb', self._injector.inject_condition_expressions, unique_id='dynamodb-condition-expression', ) # Apply the handler that serializes the request from python # types to dynamodb types. self.meta.client.meta.events.register( 'before-parameter-build.dynamodb', self._injector.inject_attribute_value_input, unique_id='dynamodb-attr-value-input', ) # Apply the handler that deserializes the response from dynamodb # types to python types. self.meta.client.meta.events.register( 'after-call.dynamodb', self._injector.inject_attribute_value_output, unique_id='dynamodb-attr-value-output', ) # Apply the documentation customizations to account for # the transformations. attr_value_shape_docs = DocumentModifiedShape( 'AttributeValue', new_type='valid DynamoDB type', new_description=( '- The value of the attribute. The valid value types are ' 'listed in the ' ':ref:`DynamoDB Reference Guide<ref_valid_dynamodb_types>`.' ), new_example_value=( '\'string\'\|123\|Binary(b\'bytes\')\|True\|None\|set([\'string\'])' '\|set([123])\|set([Binary(b\'bytes\')])\|[]\|{}' ), ) key_expression_shape_docs = DocumentModifiedShape( 'KeyExpression', new_type=( 'condition from :py:class:`boto3.dynamodb.conditions.Key` ' 'method' ), new_description=( 'The condition(s) a key(s) must meet. Valid conditions are ' 'listed in the ' ':ref:`DynamoDB Reference Guide<ref_dynamodb_conditions>`.' ), new_example_value='Key(\'mykey\').eq(\'myvalue\')', ) con_expression_shape_docs = DocumentModifiedShape( 'ConditionExpression', new_type=( 'condition from :py:class:`boto3.dynamodb.conditions.Attr` ' 'method' ), new_description=( 'The condition(s) an attribute(s) must meet. Valid conditions ' 'are listed in the ' ':ref:`DynamoDB Reference Guide<ref_dynamodb_conditions>`.' ), new_example_value='Attr(\'myattribute\').eq(\'myvalue\')', ) self.meta.client.meta.events.register( 'docs..dynamodb..complete-section', attr_value_shape_docs.replace_documentation_for_matching_shape, unique_id='dynamodb-attr-value-docs', ) self.meta.client.meta.events.register( 'docs..dynamodb..complete-section', key_expression_shape_docs.replace_documentation_for_matching_shape, unique_id='dynamodb-key-expression-docs', ) self.meta.client.meta.events.register( 'docs..dynamodb..complete-section', con_expression_shape_docs.replace_documentation_for_matching_shape, unique_id='dynamodb-cond-expression-docs', ) class TransformationInjector: """Injects the transformations into the user provided parameters.""" def __init__( self, transformer=None, condition_builder=None, serializer=None, deserializer=None, ): self._transformer = transformer if transformer is None: self._transformer = ParameterTransformer() self._condition_builder = condition_builder if condition_builder is None: self._condition_builder = ConditionExpressionBuilder() self._serializer = serializer if serializer is None: self._serializer = TypeSerializer() self._deserializer = deserializer if deserializer is None: self._deserializer = TypeDeserializer() def inject_condition_expressions(self, params, model, kwargs): """Injects the condition expression transformation into the parameters This injection includes transformations for ConditionExpression shapes and KeyExpression shapes. It also handles any placeholder names and values that are generated when transforming the condition expressions. """ self._condition_builder.reset() generated_names = {} generated_values = {} # Create and apply the Condition Expression transformation. transformation = ConditionExpressionTransformation( self._condition_builder, placeholder_names=generated_names, placeholder_values=generated_values, is_key_condition=False, ) self._transformer.transform( params, model.input_shape, transformation, 'ConditionExpression' ) # Create and apply the Key Condition Expression transformation. transformation = ConditionExpressionTransformation( self._condition_builder, placeholder_names=generated_names, placeholder_values=generated_values, is_key_condition=True, ) self._transformer.transform( params, model.input_shape, transformation, 'KeyExpression' ) expr_attr_names_input = 'ExpressionAttributeNames' expr_attr_values_input = 'ExpressionAttributeValues' # Now that all of the condition expression transformation are done, # update the placeholder dictionaries in the request. if expr_attr_names_input in params: params[expr_attr_names_input].update(generated_names) else: if generated_names: params[expr_attr_names_input] = generated_names if expr_attr_values_input in params: params[expr_attr_values_input].update(generated_values) else: if generated_values: params[expr_attr_values_input] = generated_values def inject_attribute_value_input(self, params, model, kwargs): """Injects DynamoDB serialization into parameter input""" self._transformer.transform( params, model.input_shape, self._serializer.serialize, 'AttributeValue', ) def inject_attribute_value_output(self, parsed, model, *kwargs): """Injects DynamoDB deserialization into responses""" if model.output_shape is not None: self._transformer.transform( parsed, model.output_shape, self._deserializer.deserialize, 'AttributeValue', ) class ConditionExpressionTransformation: """Provides a transformation for condition expressions The ``ParameterTransformer`` class can call this class directly to transform the condition expressions in the parameters provided. """ def __init__( self, condition_builder, placeholder_names, placeholder_values, is_key_condition=False, ): self._condition_builder = condition_builder self._placeholder_names = placeholder_names self._placeholder_values = placeholder_values self._is_key_condition = is_key_condition def __call__(self, value): if isinstance(value, ConditionBase): # Create a conditional expression string with placeholders # for the provided condition. built_expression = self._condition_builder.build_expression( value, is_key_condition=self._is_key_condition ) self._placeholder_names.update( built_expression.attribute_name_placeholders ) self._placeholder_values.update( built_expression.attribute_value_placeholders ) return built_expression.condition_expression # Use the user provided value if it is not a ConditonBase object. return value class ParameterTransformer: """Transforms the input to and output from botocore based on shape""" def transform(self, params, model, transformation, target_shape): """Transforms the dynamodb input to or output from botocore It applies a specified transformation whenever a specific shape name is encountered while traversing the parameters in the dictionary. :param params: The parameters structure to transform. :param model: The operation model. :param transformation: The function to apply the parameter :param target_shape: The name of the shape to apply the transformation to """ self._transform_parameters(model, params, transformation, target_shape) def _transform_parameters( self, model, params, transformation, target_shape ): type_name = model.type_name if type_name in ('structure', 'map', 'list'): getattr(self, f'_transform_{type_name}')( model, params, transformation, target_shape ) def _transform_structure( self, model, params, transformation, target_shape ): if not isinstance(params, collections_abc.Mapping): return for param in params: if param in model.members: member_model = model.members[param] member_shape = member_model.name if member_shape == target_shape: params[param] = transformation(params[param]) else: self._transform_parameters( member_model, params[param], transformation, target_shape, ) def _transform_map(self, model, params, transformation, target_shape): if not isinstance(params, collections_abc.Mapping): return value_model = model.value value_shape = value_model.name for key, value in params.items(): if value_shape == target_shape: params[key] = transformation(value) else: self._transform_parameters( value_model, params[key], transformation, target_shape ) def _transform_list(self, model, params, transformation, target_shape): if not isinstance(params, collections_abc.MutableSequence): return member_model = model.member member_shape = member_model.name for i, item in enumerate(params): if member_shape == target_shape: params[i] = transformation(item) else: self._transform_parameters( member_model, params[i], transformation, target_shape )	12.61 KB	2025-05-25 12:26
📄 types.py # Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. from decimal import ( Clamped, Context, Decimal, Inexact, Overflow, Rounded, Underflow, ) from boto3.compat import collections_abc STRING = 'S' NUMBER = 'N' BINARY = 'B' STRING_SET = 'SS' NUMBER_SET = 'NS' BINARY_SET = 'BS' NULL = 'NULL' BOOLEAN = 'BOOL' MAP = 'M' LIST = 'L' DYNAMODB_CONTEXT = Context( Emin=-128, Emax=126, prec=38, traps=[Clamped, Overflow, Inexact, Rounded, Underflow], ) BINARY_TYPES = (bytearray, bytes) class Binary: """A class for representing Binary in dynamodb Especially for Python 2, use this class to explicitly specify binary data for item in DynamoDB. It is essentially a wrapper around binary. Unicode and Python 3 string types are not allowed. """ def __init__(self, value): if not isinstance(value, BINARY_TYPES): types = ', '.join([str(t) for t in BINARY_TYPES]) raise TypeError(f'Value must be of the following types: {types}') self.value = value def __eq__(self, other): if isinstance(other, Binary): return self.value == other.value return self.value == other def __ne__(self, other): return not self.__eq__(other) def __repr__(self): return f'Binary({self.value!r})' def __str__(self): return self.value def __bytes__(self): return self.value def __hash__(self): return hash(self.value) class TypeSerializer: """This class serializes Python data types to DynamoDB types.""" def serialize(self, value): """The method to serialize the Python data types. :param value: A python value to be serialized to DynamoDB. Here are the various conversions: Python DynamoDB ------ -------- None {'NULL': True} True/False {'BOOL': True/False} int/Decimal {'N': str(value)} string {'S': string} Binary/bytearray/bytes (py3 only) {'B': bytes} set([int/Decimal]) {'NS': [str(value)]} set([string]) {'SS': [string]) set([Binary/bytearray/bytes]) {'BS': [bytes]} list {'L': list} dict {'M': dict} For types that involve numbers, it is recommended that ``Decimal`` objects are used to be able to round-trip the Python type. For types that involve binary, it is recommended that ``Binary`` objects are used to be able to round-trip the Python type. :rtype: dict :returns: A dictionary that represents a dynamoDB data type. These dictionaries can be directly passed to botocore methods. """ dynamodb_type = self._get_dynamodb_type(value) serializer = getattr(self, f'_serialize_{dynamodb_type}'.lower()) return {dynamodb_type: serializer(value)} def _get_dynamodb_type(self, value): dynamodb_type = None if self._is_null(value): dynamodb_type = NULL elif self._is_boolean(value): dynamodb_type = BOOLEAN elif self._is_number(value): dynamodb_type = NUMBER elif self._is_string(value): dynamodb_type = STRING elif self._is_binary(value): dynamodb_type = BINARY elif self._is_type_set(value, self._is_number): dynamodb_type = NUMBER_SET elif self._is_type_set(value, self._is_string): dynamodb_type = STRING_SET elif self._is_type_set(value, self._is_binary): dynamodb_type = BINARY_SET elif self._is_map(value): dynamodb_type = MAP elif self._is_listlike(value): dynamodb_type = LIST else: msg = f'Unsupported type "{type(value)}" for value "{value}"' raise TypeError(msg) return dynamodb_type def _is_null(self, value): if value is None: return True return False def _is_boolean(self, value): if isinstance(value, bool): return True return False def _is_number(self, value): if isinstance(value, (int, Decimal)): return True elif isinstance(value, float): raise TypeError( 'Float types are not supported. Use Decimal types instead.' ) return False def _is_string(self, value): if isinstance(value, str): return True return False def _is_binary(self, value): if isinstance(value, (Binary, bytearray, bytes)): return True return False def _is_set(self, value): if isinstance(value, collections_abc.Set): return True return False def _is_type_set(self, value, type_validator): if self._is_set(value): if False not in map(type_validator, value): return True return False def _is_map(self, value): if isinstance(value, collections_abc.Mapping): return True return False def _is_listlike(self, value): if isinstance(value, (list, tuple)): return True return False def _serialize_null(self, value): return True def _serialize_bool(self, value): return value def _serialize_n(self, value): number = str(DYNAMODB_CONTEXT.create_decimal(value)) if number in ['Infinity', 'NaN']: raise TypeError('Infinity and NaN not supported') return number def _serialize_s(self, value): return value def _serialize_b(self, value): if isinstance(value, Binary): value = value.value return value def _serialize_ss(self, value): return [self._serialize_s(s) for s in value] def _serialize_ns(self, value): return [self._serialize_n(n) for n in value] def _serialize_bs(self, value): return [self._serialize_b(b) for b in value] def _serialize_l(self, value): return [self.serialize(v) for v in value] def _serialize_m(self, value): return {k: self.serialize(v) for k, v in value.items()} class TypeDeserializer: """This class deserializes DynamoDB types to Python types.""" def deserialize(self, value): """The method to deserialize the DynamoDB data types. :param value: A DynamoDB value to be deserialized to a pythonic value. Here are the various conversions: DynamoDB Python -------- ------ {'NULL': True} None {'BOOL': True/False} True/False {'N': str(value)} Decimal(str(value)) {'S': string} string {'B': bytes} Binary(bytes) {'NS': [str(value)]} set([Decimal(str(value))]) {'SS': [string]} set([string]) {'BS': [bytes]} set([bytes]) {'L': list} list {'M': dict} dict :returns: The pythonic value of the DynamoDB type. """ if not value: raise TypeError( 'Value must be a nonempty dictionary whose key ' 'is a valid dynamodb type.' ) dynamodb_type = list(value.keys())[0] try: deserializer = getattr( self, f'_deserialize_{dynamodb_type}'.lower() ) except AttributeError: raise TypeError(f'Dynamodb type {dynamodb_type} is not supported') return deserializer(value[dynamodb_type]) def _deserialize_null(self, value): return None def _deserialize_bool(self, value): return value def _deserialize_n(self, value): return DYNAMODB_CONTEXT.create_decimal(value) def _deserialize_s(self, value): return value def _deserialize_b(self, value): return Binary(value) def _deserialize_ns(self, value): return set(map(self._deserialize_n, value)) def _deserialize_ss(self, value): return set(map(self._deserialize_s, value)) def _deserialize_bs(self, value): return set(map(self._deserialize_b, value)) def _deserialize_l(self, value): return [self.deserialize(v) for v in value] def _deserialize_m(self, value): return {k: self.deserialize(v) for k, v in value.items()}	9.32 KB	2025-05-25 12:26

Terminal