WL11281: Transaction savepoints

nmariz · nmariz · commit ad67f3b56bcc · 2017-12-07T09:52:39.000Z
This patch extends transaction API to allow setting savepoints operations.

The SAVEPOINT statement sets a named transaction allowing to parts of a
transaction to be rolled back before COMMIT.
To allow these operations, this patch adds set_savepoint(), rollback_to()
and release_savepoint() methods to Session class.

Tests were added for regression.
diff --git a/docs/mysqlx/tutorials.rst b/docs/mysqlx/tutorials.rst
@@ -5,4 +5,4 @@ Tutorials
    :maxdepth: 4
 
    tutorials/getting_started
-
+   tutorials/transactions
diff --git a/docs/mysqlx/tutorials/transactions.rst b/docs/mysqlx/tutorials/transactions.rst
@@ -0,0 +1,112 @@
+Transactions
+============
+
+Savepoints
+----------
+
+The ``SAVEPOINT`` statement sets a named transaction allowing parts of a transaction to be rolled back before ``COMMIT``.
+
+Get the collection object
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Assuming the existence of ``test_schema.test_collection`` collection.
+
+.. code-block:: python
+
+   [{
+       "_id": 1,
+       "name": "Fred",
+       "age": 21
+   }]
+
+
+Get the collection object.
+
+.. code-block:: python
+
+   session = mysqlx.get_session("root:@localhost:33060")
+   schema = session.get_schema("test_schema")
+   collection = schema.get_collection("test_collection")
+
+Set and rollback to a named transaction savepoint
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+A savepoint name can be provided to create a transaction savepoint, which can later be used to rollback.
+
+A :class:`mysqlx.OperationalError` will be raised if the savepoint name is an invalid string or if a nonexistent savepoint is being used in :func:`mysqlx.Session.rollback_to`.
+
+.. code-block:: python
+
+   # Start transaction
+   session.start_transaction()
+
+   collection.add({"name": "Wilma", "age": 33}).execute()
+   assert(2 == collection.count())
+
+   # Create a savepoint
+   session.set_savepoint("sp")
+
+   collection.add({"name": "Barney", "age": 42}).execute()
+   assert(3 == collection.count())
+
+   # Rollback to a savepoint
+   session.rollback_to("sp")
+
+   assert(2 == collection.count())
+
+   # Commit all operations
+   session.commit()
+
+Set and rollback to an unnamed transaction savepoint
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+If a savepoint name is not provided, :func:`mysqlx.Session.release_savepoint` will return a generated savepoint name.
+
+.. code-block:: python
+
+   # Start transaction
+   session.start_transaction()
+
+   collection.add({"name": "Wilma", "age": 33}).execute()
+   assert(2 == collection.count())
+
+   # Create a savepoint
+   savepoint = session.set_savepoint()
+
+   collection.add({"name": "Barney", "age": 42}).execute()
+   assert(3 == collection.count())
+
+   # Rollback to a savepoint
+   session.rollback_to(savepoint)
+
+   assert(2 == collection.count())
+
+   # Commit all operations
+   session.commit()
+
+Releasing a transaction savepoint
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+A :class:`mysqlx.OperationalError` will be raised if a nonexistent savepoint is being used in :func:`mysqlx.Session.release_savepoint`.
+
+.. code-block:: python
+
+   # Start transaction
+   session.start_transaction()
+
+   collection.add({"name": "Wilma", "age": 33}).execute()
+   assert(2 == collection.count())
+
+   # Create a savepoint
+   session.set_savepoint("sp")
+
+   collection.add({"name": "Barney", "age": 42}).execute()
+   assert(3 == collection.count())
+
+   # Release a savepoint
+   session.release_savepoint("sp")
+
+   assert(3 == collection.count())
+
+   # Commit all operations
+   session.commit()
diff --git a/lib/mysqlx/connection.py b/lib/mysqlx/connection.py
@@ -32,6 +32,7 @@
 import sys
 import socket
 import logging
+import uuid
 
 from functools import wraps
 
@@ -44,7 +45,7 @@
 from .helpers import get_item_or_attr
 from .protocol import Protocol, MessageReaderWriter
 from .result import Result, RowResult, DocResult
-from .statement import SqlStatement, AddStatement
+from .statement import SqlStatement, AddStatement, quote_identifier
 
 
 _DROP_DATABASE_QUERY = "DROP DATABASE IF EXISTS `{0}`"
@@ -720,6 +721,51 @@ def rollback(self):
         """
         self._connection.execute_nonquery("sql", "ROLLBACK", True)
 
+    def set_savepoint(self, name=None):
+        """Creates a transaction savepoint.
+
+        If a name is not provided, one will be generated using the uuid.uuid1()
+        function.
+
+        Args:
+            name (Optional[string]): The savepoint name.
+
+        Returns:
+            string: The savepoint name.
+        """
+        if name is None:
+            name = "{0}".format(uuid.uuid1())
+        elif not isinstance(name, STRING_TYPES) or len(name.strip()) == 0:
+            raise ProgrammingError("Invalid SAVEPOINT name")
+        self._connection.execute_nonquery("sql", "SAVEPOINT {0}"
+                                          "".format(quote_identifier(name)),
+                                          True)
+        return name
+
+    def rollback_to(self, name):
+        """Rollback to a transaction savepoint with the given name.
+
+        Args:
+            name (string): The savepoint name.
+        """
+        if not isinstance(name, STRING_TYPES) or len(name.strip()) == 0:
+            raise ProgrammingError("Invalid SAVEPOINT name")
+        self._connection.execute_nonquery("sql", "ROLLBACK TO SAVEPOINT {0}"
+                                          "".format(quote_identifier(name)),
+                                          True)
+
+    def release_savepoint(self, name):
+        """Release a transaction savepoint with the given name.
+
+        Args:
+            name (string): The savepoint name.
+        """
+        if not isinstance(name, STRING_TYPES) or len(name.strip()) == 0:
+            raise ProgrammingError("Invalid SAVEPOINT name")
+        self._connection.execute_nonquery("sql", "RELEASE SAVEPOINT {0}"
+                                          "".format(quote_identifier(name)),
+                                          True)
+
     def close(self):
         """Closes the session."""
         self._connection.close_session()
diff --git a/lib/mysqlx/statement.py b/lib/mysqlx/statement.py
@@ -74,7 +74,9 @@ def quote_identifier(identifier, sql_mode=""):
     Returns:
         A string with the identifier quoted with backticks.
     """
-    if is_quoted_identifier(identifier, sql_mode):
+    if len(identifier) == 0:
+        return "``"
+    elif is_quoted_identifier(identifier, sql_mode):
         return identifier
     if "ANSI_QUOTES" in sql_mode:
         return '"{0}"'.format(identifier.replace('"', '""'))
diff --git a/tests/test_mysqlx_connection.py b/tests/test_mysqlx_connection.py
@@ -446,6 +446,79 @@ def test_commit(self):
 
         drop_table(schema, table_name)
 
+    def test_savepoint(self):
+        collection_name = "collection_test"
+        schema = self.session.get_schema(self.schema_name)
+
+        # The savepoint name should be a valid string
+        self.assertRaises(mysqlx.errors.ProgrammingError,
+                          self.session.set_savepoint, 123)
+
+        # The savepoint name should not be an empty string
+        self.assertRaises(mysqlx.errors.ProgrammingError,
+                          self.session.set_savepoint, "")
+
+        # The savepoint name should not be a white space
+        self.assertRaises(mysqlx.errors.ProgrammingError,
+                          self.session.set_savepoint, " ")
+
+        # Invalid rollback savepoint without a started transaction
+        sp1 = self.session.set_savepoint("sp1")
+        self.assertRaises(mysqlx.errors.OperationalError,
+                          self.session.rollback_to, sp1)
+
+        collection = schema.create_collection(collection_name)
+
+        self.session.start_transaction()
+
+        collection.add({"name": "Fred", "age": 21}).execute()
+        self.assertEqual(1, collection.count())
+
+        # Create a savepoint named 'sp2'
+        sp2 = self.session.set_savepoint("sp2")
+        self.assertEqual(sp2, "sp2")
+
+        collection.add({"name": "Wilma", "age": 33}).execute()
+        self.assertEqual(2, collection.count())
+
+        # Create a savepoint named 'sp3'
+        sp3 = self.session.set_savepoint("sp3")
+
+        collection.add({"name": "Betty", "age": 67}).execute()
+        self.assertEqual(3, collection.count())
+
+        # Rollback to 'sp3' savepoint
+        self.session.rollback_to(sp3)
+        self.assertEqual(2, collection.count())
+
+        # Rollback to 'sp2' savepoint
+        self.session.rollback_to(sp2)
+        self.assertEqual(1, collection.count())
+
+        # The 'sp3' savepoint should not exist at this point
+        self.assertRaises(mysqlx.errors.OperationalError,
+                          self.session.rollback_to, sp3)
+
+        collection.add({"name": "Barney", "age": 42}).execute()
+        self.assertEqual(2, collection.count())
+
+        # Create an unnamed savepoint
+        sp4 = self.session.set_savepoint()
+
+        collection.add({"name": "Wilma", "age": 33}).execute()
+        self.assertEqual(3, collection.count())
+
+        # Release unnamed savepoint
+        self.session.release_savepoint(sp4)
+        self.assertEqual(3, collection.count())
+
+        # The 'sp4' savepoint should not exist at this point
+        self.assertRaises(mysqlx.errors.OperationalError,
+                          self.session.rollback_to, sp4)
+
+        self.session.commit()
+        schema.drop_collection(collection_name)
+
     def test_close(self):
         session = mysqlx.get_session(self.connect_kwargs)
         schema = session.get_schema(self.schema_name)
