Snippets for mongoengine upgrade blog post

all_in_one.py

from mongoengine import *

class UpdatableDocument(Document):
	
	schema_version = IntField(default=0)

	schema_updates = {} # Dictionary to hold all the updates

	def __init__(self, *args, **kwargs):
		values = self.schema_update(kwargs)
		super().__init__(*args, **values)

	def schema_update(values):
		version = values.get('schema_version', 0)
		next_version = version + 1

		# Get the n+1 version and perform the delta
		# keep doing until there are no new versions
		while self.schema_updates.get(next_version):
			update = self.schema_updates[next_version]
			update(values)
			values['schema_version'] = next_version
			next_version += 1

		return values
  
class IoTProduct(Document):
	""" Information about all devices of a specific type """
	tags = ListField(StringField())
	field_d = IntField()
  
  
def schema_update_add_product_tags(values):
	"""Adds the tags from the products to the tags field on the device

	Args:
		values (dict): A dictionary loaded from the mongodb BSON.
	Returns: None
	"""
	product = IoTProduct.objects.get(id=values['field_c'].id)
	values['tags'] = product.tags

class IoTDevice(UpdatableDocument):
	""" Information about a specific IoT device """
	field_a = StringField()
	field_b = IntField()
	field_c = ReferenceField(IoTProduct)
	tags = ListField(StringField())

	schema_updates = {
		1: schema_update_add_product_tags,
	}

example.bson

{
	"_id":  ObjectId("58d5b3000000000000000000"),
	"field_a": "Tempurature Sensor",
	"field_b": "21",
	"field_c": ObjectId("58d5b3011111111111111111")
}

initial_schemas.py

class IoTProduct(Document):
	""" Information about all devices of a specific type """
	tags = ListField(StringField())
	field_d = IntField()

class IoTDevice(Document):
	""" Information about a specific IoT device """
	field_a = StringField()
	field_b = IntField()
	field_c = ReferenceField(IoTProduct) # Reference to a product

schema_with_update.py

class IoTDevice(UpdatableDocument):
	""" Information about a specific IoT device """
	field_a = StringField()
	field_b = IntField()
	field_c = ReferenceField(IoTProduct)
	tags = ListField(StringField())

	schema_updates = {
		1: schema_update_add_product_tags,
	}

update_delta.py

def schema_update_add_product_tags(values):
	"""Adds the tags from the products to the tags field on the device

	Args:
		values (dict): A dictionary loaded from the mongodb BSON.
	Returns: None
	"""
	product = IoTProduct.objects.get(id=values['field_c'].id)
	values['tags'] = product.tags

update_interceptor.py

class UpdatableDocument(Document):
	
	schema_version = IntField(default=0)

	schema_updates = {} # Dictionary to hold all the updates

	def __init__(self, *args, **kwargs):
		values = self.schema_update(kwargs)
		super().__init__(*args, **values)

	def schema_update(values):
		version = values.get('schema_version', 0)
		next_version = version + 1

		# Get the n+1 version and perform the delta
		# keep doing until there are no new versions
		while self.schema_updates.get(next_version):
			update = self.schema_updates[next_version]
			update(values)
			values['schema_version'] = next_version
			next_version += 1

		return values

updated_schema.py

class IoTDevice(Document):
	""" Information about a specific IoT device """
	field_a = StringField()
	field_b = IntField()
	field_c = ReferenceField(IoTProduct) # Reference to a product
	tags = ListField(StringField()) # !New! field