Add the restriction of _id modification in mysqlx.DbDoc.

nmariz · nmariz · commit 988e50b2354c · 2018-03-15T18:51:20.000Z
This patch adds the restriction of _id modification in mysqlx.DbDoc.

The new method copy() returns a new copy of a mysqlx.DbDoc object
containing the doc_id provided. If doc_id is not provided, it will
be removed from the new mysqlx.DbDoc object.

A new test was created for mysqlx.DbDoc.
diff --git a/lib/mysqlx/crud.py b/lib/mysqlx/crud.py
@@ -400,8 +400,7 @@ def add_or_replace_one(self, doc_id, doc):
         """
         if not isinstance(doc, DbDoc):
             doc = DbDoc(doc)
-        doc["_id"] = doc_id
-        return self.add(doc).upsert(True).execute()
+        return self.add(doc.copy(doc_id)).upsert(True).execute()
 
     def get_one(self, doc_id):
         """Returns a Document matching the Document ID.
diff --git a/lib/mysqlx/dbdoc.py b/lib/mysqlx/dbdoc.py
@@ -54,12 +54,35 @@ def __init__(self, value):
     def __str__(self):
         return json.dumps(self.__dict__)
 
+    def __repr__(self):
+        return repr(self.__dict__)
+
     def __setitem__(self, index, value):
+        if index == "_id":
+            raise ProgrammingError("Cannot modify _id")
         self.__dict__[index] = value
 
     def __getitem__(self, index):
         return self.__dict__[index]
 
+    def copy(self, doc_id=None):
+        """Returns a new copy of a :class:`mysqlx.DbDoc` object containing the
+        `doc_id` provided. If `doc_id` is not provided, it will be removed from
+        new :class:`mysqlx.DbDoc` object.
+
+        Args:
+            doc_id (Optional[str]): Document ID
+
+        Returns:
+            mysqlx.DbDoc: A new instance of DbDoc containing the _id provided
+        """
+        new_dict = self.__dict__.copy()
+        if doc_id:
+            new_dict["_id"] = doc_id
+        elif "_id" in new_dict:
+            del new_dict["_id"]
+        return DbDoc(new_dict)
+
     def keys(self):
         """Returns the keys.
 
diff --git a/tests/test_mysqlx_crud.py b/tests/test_mysqlx_crud.py
@@ -68,6 +68,48 @@ def drop_view(schema, view_name):
     schema.get_session().sql(query).execute()
 
 
+@unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 12), "XPlugin not compatible")
+class MySQLxDbDocTests(tests.MySQLxTests):
+
+    def setUp(self):
+        self.connect_kwargs = tests.get_mysqlx_config()
+        self.schema_name = self.connect_kwargs["schema"]
+        self.collection_name = "collection_test"
+        try:
+            self.session = mysqlx.get_session(self.connect_kwargs)
+        except mysqlx.Error as err:
+            self.fail("{0}".format(err))
+        self.schema = self.session.get_schema(self.schema_name)
+        self.collection = self.schema.create_collection(self.collection_name)
+
+    def tearDown(self):
+        self.schema.drop_collection(self.collection_name)
+        self.session.close()
+
+    def test_dbdoc_creation(self):
+        doc_1 = mysqlx.DbDoc({"_id": "1", "name": "Fred", "age": 21})
+        self.collection.add(doc_1).execute()
+        self.assertEqual(1, self.collection.count())
+
+        # Don't allow _id assignment
+        self.assertRaises(mysqlx.ProgrammingError,
+                          doc_1.__setitem__, "_id", "1")
+
+        doc_2 = {"_id": "2", "name": "Wilma", "age": 33}
+        self.collection.add(doc_2).execute()
+        self.assertEqual(2, self.collection.count())
+
+        # Copying a DbDoc
+        doc_3 = self.collection.find().execute().fetch_one()
+        doc_4 = doc_3.copy("new_id")
+        self.assertEqual(doc_4["_id"], "new_id")
+        self.assertNotEqual(doc_3, doc_4)
+
+        # Copying a DbDoc without _id
+        doc_5 = mysqlx.DbDoc({"name": "Fred", "age": 21})
+        doc_6 = doc_5.copy()
+
+
 @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 12), "XPlugin not compatible")
 class MySQLxSchemaTests(tests.MySQLxTests):
 
