How to use pynamodb with CRUD

Pynamodb is a python library that is very helpful to play with dynamodb. It even helps us to do with a shorter code than ruby. Now let’s go through CRUD operations in short.

Import class and data type. Declaring Meta class which is the nested class containing the configuration for pynamodb such as table_name and region. Additional configs please see in source code of library.

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, NumberAttribute, JSONAttribute


class MonitoringLog(Model):
    class Meta:
        table_name = 'monitoring-log'
        region = 'ap-southeast-1'

    client_id = NumberAttribute(hash_key=True)
    date = UnicodeAttribute(range_key=True)
    summary = JSONAttribute()

Hash_key is the partition key, range_key is the sort key.

Create

Providing partition key, sort key and additional column like summary.

MonitoringLog(258, '16012024', summary={'test': 'test'}).save()

Read

Easily doing this with get method.

retrieved_item = MonitoringLog.get(258, '16012024')

Update

retrieved_item = MonitoringLog.get(258, '16012024')

retrieved_item.update(
    actions=[
        MonitoringLog.summary.set({'greeting': 'hello'})
    ]
)

Delete

retrieved_item.delete()

Bonus

Query in range.

class SensorData(Model):
    class Meta:
        table_name = 'sensor_data'
        region = 'ap-southeast-1'

    sensor_id = UnicodeAttribute(hash_key=True)
    sampled_time = NumberAttribute(range_key=True)
    val = NumberAttribute()

    @staticmethod
    def query_consumption(sensor_id, start_time, end_time):
        query_result = SensorData.query(sensor_id, SensorData.sampled_time.between(start_time, end_time))
        for item in query_result:
            print(f"Item ID: {item.sensor_id}, Timestamp: {item.sampled_time}, Value: {item.val}")

Query One

@staticmethod
    def query_one(sensor_id, start_time, end_time, asc=True):
        query_result = SensorData.query(
            sensor_id,
            SensorData.sampled_time.between(start_time, end_time),
            limit=1,
            scan_index_forward=asc
        )
        return next(query_result, None)

next(query_result_iterator, None) retrieves the first item from the iterator. If there are no items, it returns None.

Other operations please check on the document page.