in src/dynamodb_encryption_sdk/internal/formatting/serialize/attribute.py [0:0]
def serialize_attribute(attribute): # noqa: C901 pylint: disable=too-many-locals
# type: (dynamodb_types.RAW_ATTRIBUTE) -> bytes
"""Serializes a raw attribute to a byte string as defined for the DynamoDB Client-Side Encryption Standard.
:param dict attribute: Item attribute value
:returns: Serialized attribute
:rtype: bytes
"""
def _transform_binary_value(value):
# type: (dynamodb_types.BINARY) -> bytes
"""
:param value: Input value
:type value: boto3.dynamodb.types.Binary
:returns: bytes value
:rtype: bytes
"""
if isinstance(value, Binary):
return bytes(value.value)
return bytes(value)
def _serialize_binary(_attribute):
# type: (dynamodb_types.BINARY) -> bytes
# pylint: disable=no-member
# for some reason pylint can't follow the Enum member attributes
"""
:param _attribute: Attribute to serialize
:type _attribute: boto3.dynamodb.types.Binary
:returns: Serialized _attribute
:rtype: bytes
"""
return _RESERVED + Tag.BINARY.tag + encode_value(_transform_binary_value(_attribute))
def _transform_number_value(value):
# type: (str) -> bytes
"""
:param value: Input value
:type value: numbers.Number
:returns: bytes value
:rtype: bytes
"""
# At this point we are receiving values which have already been transformed
# by dynamodb.TypeSerializer, so all numbers are str. However, TypeSerializer
# leaves trailing zeros if they are defined in the Decimal call, but we need to
# strip all trailing zeros.
decimal_value = DYNAMODB_CONTEXT.create_decimal(value).normalize()
return "{0:f}".format(decimal_value).encode("utf-8")
def _serialize_number(_attribute):
# type: (str) -> bytes
# pylint: disable=no-member
# for some reason pylint can't follow the Enum member attributes
"""
:param _attribute: Attribute to serialize
:type _attribute: numbers.Number
:returns: Serialized _attribute
:rtype: bytes
"""
return _RESERVED + Tag.NUMBER.tag + encode_value(_transform_number_value(_attribute))
def _transform_string_value(value):
# type: (dynamodb_types.STRING) -> bytes
"""
:param value: Input value
:type value: bytes or str
:returns: bytes value
:rtype: bytes
"""
return to_bytes(value)
def _serialize_string(_attribute):
# type: (dynamodb_types.STRING) -> bytes
# pylint: disable=no-member
# for some reason pylint can't follow the Enum member attributes
"""
:param _attribute: Attribute to serialize
:type _attribute: six.string_types
:returns: Serialized _attribute
:rtype: bytes
"""
return _RESERVED + Tag.STRING.tag + encode_value(_transform_string_value(_attribute))
def _serialize_boolean(_attribute):
# type: (dynamodb_types.BOOLEAN) -> bytes
# pylint: disable=no-member
# for some reason pylint can't follow the Enum member attributes
"""
:param bool _attribute: Attribute to serialize
:returns: Serialized _attribute
:rtype: bytes
"""
_attribute_value = TagValues.TRUE.value if _attribute else TagValues.FALSE.value
return _RESERVED + Tag.BOOLEAN.tag + _attribute_value
def _serialize_null(_attribute):
# type: (dynamodb_types.NULL) -> bytes
# pylint: disable=no-member
# for some reason pylint can't follow the Enum member attributes
"""
:param _attribute: Attribute to serialize
:type _attribute: types.NoneType
:returns: Serialized _attribute
:rtype: bytes
"""
return _RESERVED + Tag.NULL.tag
def _serialize_set(tag, _attribute, member_function):
# type: (Tag, dynamodb_types.SET[dynamodb_types.ATTRIBUTE], Callable) -> bytes
"""
:param bytes tag: Tag to identify this set
:param set _attribute: Attribute to serialize
:param member_function: Serialization function for members
:returns: Serialized _attribute
:rtype: bytes
"""
serialized_attribute = io.BytesIO()
serialized_attribute.write(_RESERVED)
serialized_attribute.write(tag.tag)
serialized_attribute.write(encode_length(_attribute))
encoded_members = []
for member in _attribute:
encoded_members.append(member_function(member))
for member in sorted(encoded_members):
serialized_attribute.write(encode_value(member))
return serialized_attribute.getvalue()
def _serialize_binary_set(_attribute):
# type: (dynamodb_types.SET[dynamodb_types.ATTRIBUTE]) -> bytes
"""
:param set _attribute: Attribute to serialize
:returns: Serialized _attribute
:rtype: bytes
"""
return _serialize_set(Tag.BINARY_SET, _attribute, _transform_binary_value)
def _serialize_number_set(_attribute):
# type: (dynamodb_types.SET[dynamodb_types.ATTRIBUTE]) -> bytes
"""
:param set _attribute: Attribute to serialize
:returns: Serialized _attribute
:rtype: bytes
"""
return _serialize_set(Tag.NUMBER_SET, _attribute, _transform_number_value)
def _serialize_string_set(_attribute):
# type: (dynamodb_types.SET[dynamodb_types.ATTRIBUTE]) -> bytes
"""
:param set _attribute: Attribute to serialize
:returns: Serialized _attribute
:rtype: bytes
"""
return _serialize_set(Tag.STRING_SET, _attribute, _transform_string_value)
def _serialize_list(_attribute):
# type: (dynamodb_types.LIST) -> bytes
# pylint: disable=no-member
# for some reason pylint can't follow the Enum member attributes
"""
:param list _attribute: Attribute to serialize
:returns: Serialized _attribute
:rtype: bytes
"""
serialized_attribute = io.BytesIO()
serialized_attribute.write(_RESERVED)
serialized_attribute.write(Tag.LIST.tag)
serialized_attribute.write(encode_length(_attribute))
for member in _attribute:
serialized_attribute.write(serialize_attribute(member))
return serialized_attribute.getvalue()
def _serialize_map(_attribute):
# type: (dynamodb_types.MAP) -> bytes
"""
:param list _attribute: Attribute to serialize
:returns: Serialized _attribute
:rtype: bytes
"""
serialized_attribute = io.BytesIO()
serialized_attribute.write(_RESERVED)
# for some reason pylint can't follow the Enum member attributes
serialized_attribute.write(Tag.MAP.tag) # pylint: disable=no-member
serialized_attribute.write(encode_length(_attribute))
sorted_items = _sorted_key_map(item=_attribute, transform=_transform_string_value)
for key, value, _original_key in sorted_items:
serialized_attribute.write(_serialize_string(key))
serialized_attribute.write(serialize_attribute(value))
return serialized_attribute.getvalue()
def _serialize_function(dynamodb_tag):
# type: (str) -> Callable[[dynamodb_types.ATTRIBUTE], bytes]
"""Locates the appropriate serialization function for the specified DynamoDB attribute tag."""
# for some reason pylint can't follow the Enum member attributes
serialize_functions = {
Tag.BINARY.dynamodb_tag: _serialize_binary, # pylint: disable=no-member
Tag.BINARY_SET.dynamodb_tag: _serialize_binary_set, # pylint: disable=no-member
Tag.NUMBER.dynamodb_tag: _serialize_number, # pylint: disable=no-member
Tag.NUMBER_SET.dynamodb_tag: _serialize_number_set, # pylint: disable=no-member
Tag.STRING.dynamodb_tag: _serialize_string, # pylint: disable=no-member
Tag.STRING_SET.dynamodb_tag: _serialize_string_set, # pylint: disable=no-member
Tag.BOOLEAN.dynamodb_tag: _serialize_boolean, # pylint: disable=no-member
Tag.NULL.dynamodb_tag: _serialize_null, # pylint: disable=no-member
Tag.LIST.dynamodb_tag: _serialize_list, # pylint: disable=no-member
Tag.MAP.dynamodb_tag: _serialize_map, # pylint: disable=no-member
}
try:
return serialize_functions[dynamodb_tag]
except KeyError:
raise SerializationError('Unsupported DynamoDB data type: "{}"'.format(dynamodb_tag))
if not isinstance(attribute, dict):
raise TypeError('Invalid attribute type "{}": must be dict'.format(type(attribute)))
if len(attribute) != 1:
raise SerializationError(
"cannot serialize attribute: incorrect number of members {} != 1".format(len(attribute))
)
key, value = list(attribute.items())[0]
return _serialize_function(key)(value)