For our second example, we are going to use a more complex use case on a transaction with three different collections.

We are going to simulate a shopping cart and payment transaction process for an e-commerce application using MongoDB. Using the sample code that we'll provide at the end of this section, we will initially populate the database with the following data.

Our first collection is the users collection with one document per user:

> db.users.find()
{ "_id" : ObjectId("5bc22f35f8d89f2b9e01d0fd"), "user_id" : 1, "name" : "alex" }
{ "_id" : ObjectId("5bc22f35f8d89f2b9e01d0fe"), "user_id" : 2, "name" : "barbara" }

Then we have the carts collection with one document per cart, which is linked via the user_id to our users:

> db.carts.find()
{ "_id" : ObjectId("5bc2f9de8e72b42f77a20ac8"), "cart_id" : 1, "user_id" : 1 }
{ "_id" : ObjectId("5bc2f9de8e72b42f77a20ac9"), "cart_id" : 2, "user_id" : 2 }

The payments collection holds any completed payment that has gone through, storing the cart_id and the item_id to link to the cart that it belonged to and the item that has been paid:

> db.payments.find()
{ "_id" : ObjectId("5bc2f9de8e72b42f77a20aca"), "cart_id" : 1, "name" : "alex", "item_id" : 101, "status" : "paid" }

Finally, the inventories collection holds a count of the number of items (by item_id) that we have currently available, along with their price and a short description:

> db.inventories.find()
{ "_id" : ObjectId("5bc2f9de8e72b42f77a20acb"), "item_id" : 101, "description" : "bull bearing", "price" : 100, "quantity" : 5 }

In this example, we are going to demonstrate using MongoDB's schema validation functionality. Using JSON schemata, we can define a set of validations that will be checked against the database level every time a document is inserted or updated. This is a fairly new feature as it was introduced in MongoDB 3.6. In our case, we are going to use it to make sure that we always have a positive number of items in our inventory.

The validator object in the MongoDB shell format is as follows:

validator = { validator:
 { $jsonSchema:
 { bsonType: "object",
 required: ["quantity"],
 properties:
 { quantity:
 { bsonType: ["long"],
 minimum: 0,
 description: "we can’t have a negative number of items in our inventory"
 }
 }
 }
 }
}

JSON schemata can be used to implement many of the validations that we would usually have in our models in Rails or Django. We can define these keywords as in the following table:

Using JSON schema, we can offload validations from our models to the database layer and/or use MongoDB validations as an additional layer of security on top of web application validations.

To use a JSON schema, we have to specify it at the time that we are creating our collection, as follows:

> db.createCollection("inventories", validator)

Returning to our example, our code will simulate having an inventory of five bull bearings and placing two orders; one by user Alex for two bull bearings, followed by a second order by user Barbara for another four bull bearings.

As expected, the second order will not go through because we don't have enough ball bearings in our inventory to fulfill it. We will see this in the following code:

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure
from pymongo.errors import OperationFailure

class ECommerce:
   def __init__(self):
       self.client = MongoClient('localhost', 27017, w='majority')
       self.db = self.client.mongo_bank
       self.users = self.db['users']
       self.carts = self.db['carts']
       self.payments = self.db['payments']
       self.inventories = self.db['inventories']
       # delete any existing data
       self.db.drop_collection('carts')
       self.db.drop_collection('payments')
       self.db.inventories.remove()
       # insert new data
       self.insert_data()
       alex_order_cart_id = self.add_to_cart(1,101,2)
       barbara_order_cart_id = self.add_to_cart(2,101,4)
       self.place_order(alex_order_cart_id)
       self.place_order(barbara_order_cart_id)
   def insert_data(self):
       self.users.insert_one({'user_id': 1, 'name': 'alex' })
       self.users.insert_one({'user_id': 2, 'name': 'barbara'})
       self.carts.insert_one({'cart_id': 1, 'user_id': 1})
       self.db.carts.insert_one({'cart_id': 2, 'user_id': 2})
       self.db.payments.insert_one({'cart_id': 1, 'name': 'alex', 'item_id': 101, 'status': 'paid'})
       self.db.inventories.insert_one({'item_id': 101, 'description': 'bull bearing', 'price': 100, 'quantity': 5.0})

   def add_to_cart(self, user, item, quantity):
       # find cart for user
       cart_id = self.carts.find_one({'user_id':user})['cart_id']
       self.carts.update_one({'cart_id': cart_id}, {'$inc': {'quantity': quantity}, '$set': { 'item': item} })
       return cart_id

   def place_order(self, cart_id):
           while True:
               try:
                   with self.client.start_session() as ses:
                       ses.start_transaction()
                       cart = self.carts.find_one({'cart_id': cart_id}, session=ses)
                       item_id = cart['item']
                       quantity = cart['quantity']
                       # update payments
                       self.db.payments.insert_one({'cart_id': cart_id, 'item_id': item_id, 'status': 'paid'}, session=ses)
                       # remove item from cart
                       self.db.carts.update_one({'cart_id': cart_id}, {'$inc': {'quantity': quantity * (-1)}}, session=ses)
                       # update inventories
                       self.db.inventories.update_one({'item_id': item_id}, {'$inc': {'quantity': quantity*(-1)}}, session=ses)
                       ses.commit_transaction()
                       break
               except (ConnectionFailure, OperationFailure) as exc:
                   print("Transaction aborted. Caught exception during transaction.")
                   # If transient error, retry the whole transaction
                   if exc.has_error_label("TransientTransactionError"):
                       print("TransientTransactionError, retrying transaction ...")
                       continue
                   elif str(exc) == 'Document failed validation':
                       print("error validating document!")
                       raise
                   else:
                       print("Unknown error during commit ...")
                       raise
def main():
   ECommerce()
if __name__ == '__main__':
   main()

We will break down the preceding example into the interesting parts, as follows:

   def add_to_cart(self, user, item, quantity):
       # find cart for user
       cart_id = self.carts.find_one({'user_id':user})['cart_id']
       self.carts.update_one({'cart_id': cart_id}, {'$inc': {'quantity': quantity}, '$set': { 'item': item} })
       return cart_id

The add_to_cart() method doesn't use transactions. The reason is that because we are updating one document at a time, these are guaranteed to be atomic operations.

Then, in the place_order() method, we start the session, and then subsequently, a transaction within this session. Similar to the previous use case, we need to make sure that we add the session=ses parameter at the end of every operation that we want to be executed in the transaction context:

    def place_order(self, cart_id):
 while True:
 try:
 with self.client.start_session() as ses:
 ses.start_transaction()
 …
 # update payments
 self.db.payments.insert_one({'cart_id': cart_id, 'item_id': item_id, 'status': 'paid'}, session=ses)
 # remove item from cart
 self.db.carts.update_one({'cart_id': cart_id}, {'$inc': {'quantity': quantity * (-1)}}, session=ses)
 # update inventories
 self.db.inventories.update_one({'item_id': item_id}, {'$inc': {'quantity': quantity*(-1)}}, session=ses)
 ses.commit_transaction()
 break
 except (ConnectionFailure, OperationFailure) as exc:
 print("Transaction aborted. Caught exception during transaction.")
 # If transient error, retry the whole transaction
 if exc.has_error_label("TransientTransactionError"):
 print("TransientTransactionError, retrying transaction ...")
 continue
 elif str(exc) == 'Document failed validation':
 print("error validating document!")
 raise
 else:
 print("Unknown error during commit ...")
 raise

In this method, we are using the retry able transaction pattern. We start by wrapping the transaction context in a while True block, essentially making it loop forever. Then we enclose our transaction in a try block that will listen for exceptions.

An exception of type transient transaction, which has the TransientTransactionError error label, will result in continued execution in the while True block, essentially retrying the transaction from the very beginning. On the other hand, a failed validation or any other error will reraise the exception after logging it.

In the end, our database looks like the following code block:

> db.payments.find()
{ "_id" : ObjectId("5bc307178e72b431c0de385f"), "cart_id" : 1, "name" : "alex", "item_id" : 101, "status" : "paid" }
{ "_id" : ObjectId("5bc307178e72b431c0de3861"), "cart_id" : 1, "item_id" : 101, "status" : "paid" }

The payment that we just made does not have the name field, in contrast to the sample payment that we inserted in our database before rolling our transactions:

> db.inventories.find()
{ "_id" : ObjectId("5bc303468e72b43118dda074"), "item_id" : 101, "description" : "bull bearing", "price" : 100, "quantity" : 3 }

Our inventory has the correct number of bull bearings, three (five minus the two that Alex ordered), as shown in the following code block:

> db.carts.find()
{ "_id" : ObjectId("5bc307178e72b431c0de385d"), "cart_id" : 1, "user_id" : 1, "item" : 101, "quantity" : 0 }
{ "_id" : ObjectId("5bc307178e72b431c0de385e"), "cart_id" : 2, "user_id" : 2, "item" : 101, "quantity" : 4 }

Our carts have the correct quantities. Alex's cart (cart_id=1) has zero items, whereas Barbara's cart (cart_id=2) still has four, since we don't have enough bull bearings to fulfill her order. Our payments collection does not have an entry for Barbara's order and the inventory still has three bull bearings in place.

Our database state is consistent and saving lots of time by implementing the abort transaction and reconciliation data logic in our application level.

Continuing with the same example in Ruby, we have the following code block:

require 'mongo'

class ECommerce
 def initialize
   @client = Mongo::Client.new([ '127.0.0.1:27017' ], database: :mongo_bank)
   db = @client.database
   @users = db[:users]
   @carts = db[:carts]
   @payments = db[:payments]
   @inventories = db[:inventories]

   # drop any existing data
   @users.drop
   @carts.drop
   @payments.drop
   @inventories.delete_many

   # insert data
   @users.insert_one({ "user_id": 1, "name": "alex" })
   @users.insert_one({ "user_id": 2, "name": "barbara" })

   @carts.insert_one({ "cart_id": 1, "user_id": 1 })
   @carts.insert_one({ "cart_id": 2, "user_id": 2 })

   @payments.insert_one({"cart_id": 1, "name": "alex", "item_id": 101, "status": "paid" })
   @inventories.insert_one({"item_id": 101, "description": "bull bearing", "price": 100, "quantity": 5 })

   alex_order_cart_id = add_to_cart(1, 101, 2)
   barbara_order_cart_id = add_to_cart(2, 101, 4)

   place_order(alex_order_cart_id)
   place_order(barbara_order_cart_id)
 end

 def add_to_cart(user, item, quantity)
   session = @client.start_session
   session.start_transaction
   cart_id = @users.find({ "user_id": user}).first['user_id']
   @carts.update_one({"cart_id": cart_id}, {'$inc': { 'quantity': quantity }, '$set': { 'item': item } }, session: session)
   session.commit_transaction
   cart_id
 end

 def place_order(cart_id)
   session = @client.start_session
   session.start_transaction
   cart = @carts.find({'cart_id': cart_id}, session: session).first
   item_id = cart['item']
   quantity = cart['quantity']
   @payments.insert_one({'cart_id': cart_id, 'item_id': item_id, 'status': 'paid'}, session: session)
   @carts.update_one({'cart_id': cart_id}, {'$inc': {'quantity': quantity * (-1)}}, session: session)
   @inventories.update_one({'item_id': item_id}, {'$inc': {'quantity': quantity*(-1)}}, session: session)
   quantity = @inventories.find({'item_id': item_id}, session: session).first['quantity']
   if quantity < 0
     session.abort_transaction
   else
     session.commit_transaction
   end
 end
end

ECommerce.new

Similar to the Python code sample, we are passing the session: session parameter along each operation to make sure that we are operating inside the transaction.

Here, we are not using the retry able transaction pattern. Regardless, MongoDB will retry committing or aborting a transaction once before throwing an exception.