simple_dynamodb.py

import boto3
import deepdiff
import copy
import json

class SimpleTable(boto3.resources.factory.dynamodb.Table):
	def __init__(self, table, partition_key = 'pk', sort_key = None):
		self.table = table
		self.partition_key = partition_key
		self.sort_key = sort_key
	
	def update(self, record):
		return Record(self.table, record, partition_key = self.partition_key, sort_key = self.sort_key)

class Record:
	def __init__(self, table, record, partition_key, sort_key):
		self.table = table
		self.record = record
		self.snapshot = copy.deepcopy(record)
		self.index = 0

		self.update_expression_set = []
		self.update_expression_remove = []
		self.expression_attribute_names = {}
		self.expression_attribute_values = {}

		self.partition_key = partition_key
		self.sort_key = sort_key

	def __enter__(self):
		return self
	
	def __exit__(self, *args):
		diff = deepdiff.DeepDiff(self.snapshot, self.record)

		for difftype, diffinfo in diff.tree.items():
			if difftype in ('iterable_item_added', 'iterable_item_removed', 'unprocess'):
				# This is my personal hattress towards list in NoSQL since it can only be
				# removed with index, not good with parallel, tell me if you got this right
				raise RuntimeWarning()
			
			path, names = self.parse_path(change.path())
				
			if difftype in ('type_changes', 'dictionary_item_added', 'value_changed'):
				for change in diffinfo:
					varname = ':v'+str(self.index); self.index += 1
					self.update_expression_set.append('{path} = {value}'.format(
						path = path,
						value = varname
					))
			
			# Again, I hate list
			elif difftype in ('dictionary_item_removed',):
				for change in diffinfo:
					self.update_expression_remove.append('{path}'.format(
						path = path
					))
		
		# Compile time
		update_string = ''
		if len(self.update_expression_set):
			update_string += 'SET ' + ', '.join(self.update_expression_set) + ' '
		if len(self.update_expression_remove):
			update_string += 'REMOVE' + ', '.join(self.update_expression_remove) + ' '
		
		# Perform the update
		Key = {self.partition_key: self.record[self.partition_key]}
		if self.sort_key is not None:
			Key[self.sort_key] = self.record[self.sort_key]
		
		self.table.update_item(
			Key = Key,
			UpdateExpression = update_string,
			ExpressionAttributeNames = self.expression_attribute_names,
			ExpressionAttributeValues = self.expression_attribute_values
		)

	def parse_path(self, path):
		path = '#' + path[6:-2].replace("']['", ".#") #"root['nodes']['Node23']['device']"
		names = list(filter(lambda x: x.startswith('#'), path.split('.')))
		for name in names:
			self.expression_attribute_names[name] = name[1:] # Remove the #
		return path, names


if __name__ == '__main__':
	# Assuming you have setup the enviroment
	dynamodb = boto3.resource('dynamodb')
	table = dynamodb.Table('sometable')
	table = SimpleTable(table)

	# Usually, you get your item first
	item = table.get_item(Key = {"yourkey": "something"})

	# Then you want to modify it
	with table.update(item):
		item["Hello"] = "World"