Dynamo conversion

async_boto.utils.dynamo_conversion

DYNAMODB_ENCODERS module-attribute

DYNAMODB_ENCODERS = {bool: lambda v: {'BOOL': v}, str: serialize_as_string, int: serialize_as_number, float: serialize_as_number, type(None): lambda v: {'NULL': True}, list: serialize_list, tuple: serialize_list, dict: serialize_dict, set: serialize_set, Set: serialize_set, bytes: lambda b: {'B': b}, bytearray: lambda ba: {'B': bytes(ba)}, Decimal: serialize_as_number, date: lambda d: serialize_as_string(isoformat()), datetime: lambda d: serialize_as_string(isoformat()), time: lambda d: serialize_as_string(isoformat()), timedelta: lambda td: serialize_as_number(total_seconds()), Enum: lambda o: serialize_as_string(value), UUID: serialize_as_string}

DYNAMODB_TYPES module-attribute

DYNAMODB_TYPES = {'NULL': lambda v, serialize: None, 'BOOL': lambda v, serialize: v, 'N': lambda v, serialize: Decimal(v), 'S': lambda v, serialize: v, 'B': lambda v, serialize: v, 'NS': lambda v, serialize: set(Decimal(item) for item in v), 'SS': lambda v, serialize: set(v), 'BS': lambda v, serialize: set(v), 'L': lambda v, serialize: [serialize(item) for item in v], 'M': lambda v, serialize: {k_: serialize(v_)for (k_, v_) in items()}}

example_dict module-attribute

example_dict = {'id': '123', 'name': 'Item Name', 'price': 19.99, 'tags': ['tag1', 'tag2'], 'details': {'color': 'red', 'size': 'M'}, 'is_available': True, 'quantity': None}

dynamodb_json module-attribute

dynamodb_json = to_dynamodb_json(example_dict)

python_dict module-attribute

python_dict = from_dynamodb_json(dynamodb_json)

serialize_dict

serialize_dict(v, serialize)

Source code in async_boto/utils/dynamo_conversion.py

def serialize_dict(v, serialize):
    return {"M": {k_: serialize(v_) for k_, v_ in v.items()}}

serialize_list

serialize_list(v, serialize)

Source code in async_boto/utils/dynamo_conversion.py

def serialize_list(v, serialize):
    return {"L": [serialize(item) for item in v]}

serialize_set

serialize_set(v)

Source code in async_boto/utils/dynamo_conversion.py

def serialize_set(v):
    if all(isinstance(item, int | float | Decimal) for item in v):
        return {"NS": [str(item) for item in v]}
    elif all(isinstance(item, str) for item in v):
        return {"SS": [str(item) for item in v]}
    elif all(isinstance(item, bytearray | bytes) for item in v):
        return {
            "BS": [item if isinstance(item, bytes) else bytearray(item) for item in v]
        }

    raise TypeError(
        "All values in a set must be of the same family of"
        "types (numbers, strings or bytes)"
    )

serialize_as_string

serialize_as_string(v)

Source code in async_boto/utils/dynamo_conversion.py

def serialize_as_string(v):
    return {"S": str(v)}

serialize_as_number

serialize_as_number(v)

Source code in async_boto/utils/dynamo_conversion.py

def serialize_as_number(v):
    return {"N": str(v)}

decode_dynamodb_primative

decode_dynamodb_primative(value)

Source code in async_boto/utils/dynamo_conversion.py

def decode_dynamodb_primative(value):
    _type = next(iter(value))

    if _type not in DYNAMODB_TYPES:
        raise TypeError(
            f'DynamoDB does not have a type {_type}."'
            f'"Use one of {list(DYNAMODB_TYPES.keys())}'
        )

    return DYNAMODB_TYPES[_type](value[_type], decode_dynamodb_primative)

decode_dynamodb_json

decode_dynamodb_json(v)

Source code in async_boto/utils/dynamo_conversion.py

def decode_dynamodb_json(v):
    try:
        key = next(iter(v))
    except StopIteration:  # Empty dict
        return {}

    # Convert root level attribute map into {"M": ...}
    if len(v) > 1 or (len(v) == 1 and key not in DYNAMODB_TYPES):
        v = {"M": v}

    return decode_dynamodb_primative(v)

to_dynamodb_json

to_dynamodb_json(python_dict)

Convert a Python dictionary to a DynamoDB JSON format.

:param python_dict: The Python dictionary to convert. :return: A dictionary in DynamoDB JSON format.

Source code in async_boto/utils/dynamo_conversion.py

def to_dynamodb_json(python_dict):
    """
    Convert a Python dictionary to a DynamoDB JSON format.

    :param python_dict: The Python dictionary to convert.
    :return: A dictionary in DynamoDB JSON format.
    """
    if isinstance(python_dict, dict):
        dynamodb_json = {}
        for key, value in python_dict.items():
            dynamodb_json[key] = DYNAMODB_ENCODERS[type(value)](value)
        return dynamodb_json
    else:
        raise ValueError("Input must be a dictionary.")

from_dynamodb_json

from_dynamodb_json(dynamodb_json)

Convert a DynamoDB JSON format dictionary back to a Python dictionary.

:param dynamodb_json: A dictionary in DynamoDB JSON format. :return: A Python dictionary.

Source code in async_boto/utils/dynamo_conversion.py

def from_dynamodb_json(dynamodb_json):
    """
    Convert a DynamoDB JSON format dictionary back to a Python dictionary.

    :param dynamodb_json: A dictionary in DynamoDB JSON format.
    :return: A Python dictionary.
    """
    if isinstance(dynamodb_json, dict):
        return decode_dynamodb_json(dynamodb_json)
    else:
        raise ValueError("Input must be a dictionary.")