Deal with the change of field data type of PynamoDB, an ORM of DynamoDB

Effectively manage the change of field data type of PynamoDB, an object-relational mapper (ORM) for DynamoDB. Ensure smooth database schema updates and maintain application performance and integrity with this guide.

PynamoDB, with a Python interface, is a great ORM for AWS DynamoDB. DynamoDB is a NoSQL database with scalability and security and it has awesome API and SDKs. PynamoDB wraps the underneath APIs of DynamoDB and gives a beautiful pythonic way to use DynamoDB.

In real-world our DB fields are not constant. It can change with the requirements update of our project. The hard part is when you have to maintain the old and new features. For example, a field data type can change from String to Binary. So if any API e.g. REST API with versioning is working with DynamoDB, you have to accept both old data type (String) and new data type (Binary). We are going to find a solution in such a case.

Setting the goal

Let’s start with setting a goal. Assume we have a field with Dynamo data type List. But after times it changes to Map. So we will do

Accept Map object and save to DynamoDB
Accept List object and change it to Map according to requirement and save to DynamoDB as our field changed to Map
Return Map object according to requirement change

Setting the project

Let’s add a couple of files to do our task.

# Add project folder
mkdir change-field-ddb
cd change-field-ddb
# Add Pipfile for requirements mange with pipenv
touch Pipfile
# For Custom PynamoDB attributes
touch attributes.py
# Add models.py to add pynamo model
touch models.py
# To test everything
touch test.py

Add python requirements

Add following requirements to Pipfile

[dev-packages]
pipenv = "==2020.8.13"
moto = "==1.3.14"
[packages]
pynamodb = "==4.3.2"
[requires]
python_version = "3.8"

Now initiate pipenv with

pipenv install

Now the project will look like

tree
.
├── attributes.py
├── models.py
├── Pipfile
├── Pipfile.lock
└── test.py

The changed datatype

Let’s assume our previous datatype (List) for a field result is

[3.75, 3.17, 3.90, 3.67, .......]

So the result stores CGPA of a student from 1st semester to final semester e.g. result[0] as 1st semester, result[1] as 2nd semester, and so on. But we will make it better (Map) at the updated version

{
    'semester 1': 3.75,
    'semester 2': 3.17,
    'semester 3': 3.90,
    'semester 4': 3.67,
    ..................
}

PynamoDB attributes

Add the custom PynamoDB attribute to accept old (List) and new (Map) attribute and saves only as Map. But at return time it will return old (List) data as Map and also new data as Map. Also, we are going to use custom UUIDAttribute which will be used as a hash key.

# attributes.py
import uuid
import pynamodb.attributes
# UUID Attribute will use as Dynamo hash key
class UUIDAttribute(pynamodb.attributes.UnicodeAttribute):
def serialize(self, value):
return super().serialize(str(value))
def deserialize(self, value):
return uuid.UUID(super().deserialize(value))
# Add custom attribute to serialize and deserialize data
class ResultAttribute(pynamodb.attributes.MapAttribute):
@classmethod
def is_raw(cls):
# Set to use as AttributeContainer
# https://pynamodb.readthedocs.io/en/latest/api.html#pynamodb.attributes.MapAttribute
return True
@staticmethod
def _parse_value(values):
return {
f'semester {idx+1}': val for idx, val in enumerate(values)
}
def serialize(self, values):
# Convert python list to native pynamo
if isinstance(values, (list, tuple)):
values = self._parse_value(values)
return super().serialize(values)
def get_value(self, value):
try:
# Convert from
# {'L': [{'N': '3.75'}, {'N': '3.17'}]}
# to
# {'M': {'semester 1': {'N': '3.75'}, 'semester 2': {'N': '3.17'}}}
value = {'M': self._parse_value(value['L'])}
except (KeyError, TypeError):
pass
return super().get_value(value)

The PynamoDB Model

We are going to start with a simple PynamoDB Model only with 2 fields

# models.py
import uuid
import pynamodb.models
import pynamodb.attributes
from attributes import UUIDAttribute, ResultAttribute
class ResultModel(pynamodb.models.Model):
id = UUIDAttribute(hash_key=True, default=uuid.uuid4)
result = ResultAttribute()
class Meta:
table_name = "test-ddb-table"

Time to Test

Add some tests in test.py

from decimal import Decimal
from decimal import Decimal
import boto3
import moto
from models import ResultModel
with moto.mock_dynamodb2():
region = 'eu-west-1'
ResultModel.Meta.region = region
ResultModel.create_table(wait=True)
# Data
result_map = {
'semester 1': 3.75,
'semester 2': 3.17,
'semester 3': 3.90,
'semester 4': 3.67
}
result_list = [3.75, 3.17, 3.90, 3.67]
# Insert as Map with PynamoDB
result1 = ResultModel(result=result_map)
result1.save()
assert ResultModel.count() == 1
result1_id = result1.id
print(result1_id)
# Retrieve the data from inserted as list
result1_retr = ResultModel.get(result1_id)
assert result1_retr.result.attribute_values == result_map
# Insert as list so it will convert and saved as map
result2 = ResultModel(result=result_list)
result2.save()
assert ResultModel.count() == 2
result2_id = result2.id
print(result2_id)
# Retrieve the data from inserted as list
result2_retr = ResultModel.get(result2_id)
assert result2_retr.result.attribute_values == result_map
# Insert list value in result with boto3
dynamodb = boto3.resource('dynamodb', region)
table = dynamodb.Table(ResultModel.Meta.table_name)
# float is not supported by boto3. But decimal is supported. So converted to Decimal
item = [Decimal(str(v)) for v in result_list]
# Updating item instead of creating because result is MapAttribute by default
table.update_item(
Key={'id': str(result1_id)},
AttributeUpdates={
'result': {'Value': item, 'Action': 'PUT'}
}
)
assert table.get_item(Key={'id': str(result1_id)})['Item']['result'] == item
# Retrieve the data that is a list in dynamodb
result1_retr = ResultModel.get(result1_id)
assert result1_retr.result.attribute_values == result_map