diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/boto3/dynamodb')
5 files changed, 1293 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/boto3/dynamodb/__init__.py b/.venv/lib/python3.12/site-packages/boto3/dynamodb/__init__.py new file mode 100644 index 00000000..6001b27b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/dynamodb/__init__.py @@ -0,0 +1,12 @@ +# 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. diff --git a/.venv/lib/python3.12/site-packages/boto3/dynamodb/conditions.py b/.venv/lib/python3.12/site-packages/boto3/dynamodb/conditions.py new file mode 100644 index 00000000..74b3e8e7 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/dynamodb/conditions.py @@ -0,0 +1,461 @@ +# 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 diff --git a/.venv/lib/python3.12/site-packages/boto3/dynamodb/table.py b/.venv/lib/python3.12/site-packages/boto3/dynamodb/table.py new file mode 100644 index 00000000..931296bc --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/dynamodb/table.py @@ -0,0 +1,167 @@ +# 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() diff --git a/.venv/lib/python3.12/site-packages/boto3/dynamodb/transform.py b/.venv/lib/python3.12/site-packages/boto3/dynamodb/transform.py new file mode 100644 index 00000000..3944f315 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/dynamodb/transform.py @@ -0,0 +1,343 @@ +# 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 + ) diff --git a/.venv/lib/python3.12/site-packages/boto3/dynamodb/types.py b/.venv/lib/python3.12/site-packages/boto3/dynamodb/types.py new file mode 100644 index 00000000..f358b12f --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/dynamodb/types.py @@ -0,0 +1,310 @@ +# 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()} |