implemented table.update and collection.modify

updated some of the methods of result to meet spec

Author: Reggie Burnett

lib/mysqlx/__init__.py

@@ -31,7 +31,7 @@
 from .result import (ColumnMetaData, Row, Result, BufferingResult, RowResult,
                      SqlResult)
 from .statement import (Statement, FilterableStatement, SqlStatement,
-                        AddStatement, RemoveStatement, TableDeleteStatement,
+                        AddStatement, RemoveStatement, DeleteStatement,
                         CreateCollectionIndexStatement,
                         DropCollectionIndexStatement)
 from .dbdoc import DbDoc

lib/mysqlx/connection.py

@@ -104,6 +104,10 @@ def delete(self, statement):
         self.protocol.send_delete(statement)
         return Result(self)
 
+    def update(self, statement):
+        self.protocol.send_update(statement)
+        return Result(self)
+
     def execute_nonquery(self, namespace, cmd, raise_on_fail=True, *args):
         self.protocol.send_execute_statement(namespace, cmd, args)
         return Result(self)

lib/mysqlx/crud.py

@@ -21,10 +21,9 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
 
-from .statement import (AddStatement, RemoveStatement, TableDeleteStatement,
-                        FindStatement, SelectStatement,
-                        CreateCollectionIndexStatement, InsertStatement,
-                        DropCollectionIndexStatement)
+from .statement import (FindStatement, AddStatement, RemoveStatement, ModifyStatement,
+                        SelectStatement, InsertStatement, DeleteStatement, UpdateStatement,
+                        CreateCollectionIndexStatement, DropCollectionIndexStatement)
 
 
 _COUNT_TABLES_QUERY = ("SELECT COUNT(*) FROM information_schema.tables "
@@ -145,6 +144,9 @@ def remove(self, condition=None):
             rs.where(condition)
         return rs
 
+    def modify(self, condition=None):
+        return ModifyStatement(self, condition)
+
     def count(self):
         sql = _COUNT_QUERY.format(self._schema.name, self._name)
         return self._connection.execute_sql_scalar(sql)
@@ -182,10 +184,10 @@ def insert(self, *fields):
         return InsertStatement(self, *fields)
 
     def update(self):
-        pass
+        return UpdateStatement(self)
 
     def delete(self, condition=None):
-        return TableDeleteStatement(self, condition)
+        return DeleteStatement(self, condition)
 
     def count(self):
         sql = _COUNT_QUERY.format(self._schema.name, self._name)

lib/mysqlx/expr.py

@@ -653,6 +653,9 @@ def expr(self):
     def parse_table_insert_field(self):
         return Column(name=self.consume_token(TokenType.IDENT))
 
+    def parse_table_update_field(self):
+        return self.column_identifier().identifier
+
     def parse_table_select_projection(self):
         project_expr = []
         first = True

lib/mysqlx/protocol.py

@@ -140,13 +140,18 @@ def send_find(self, stmt):
         self._apply_filter(find, stmt)
         self._writer.write_message(MySQLx.ClientMessages.CRUD_FIND, find)
 
-    def send_insert(self, schema, target, is_docs, rows, cols):
-        stmt = MySQLxCrud.Insert(
-            datamodel=MySQLxCrud.DOCUMENT if is_docs else MySQLxCrud.TABLE,
-            collection=MySQLxCrud.Collection(name=target, schema=schema))
-        for row in rows:
-            typed_row = MySQLxCrud.Insert.TypedRow()
-            stmt.rows.extend(row)
+    def send_update(self, statement):
+        update = MySQLxCrud.Update(
+            data_model=MySQLxCrud.DOCUMENT if statement._doc_based else MySQLxCrud.TABLE,
+            collection=MySQLxCrud.Collection(name=statement.target.name, schema=statement.schema.name))
+        self._apply_filter(update, statement)
+        for update_op in statement._update_ops:
+            opexpr = UpdateOperation(operation=update_op.update_type, source=update_op.source)
+            if update_op.value != None:
+                opexpr.value.CopyFrom(self.arg_object_to_expr(update_op.value, not statement._doc_based))
+            update.operation.extend([opexpr])
+        self._writer.write_message(MySQLx.ClientMessages.CRUD_UPDATE, update)
+
 
     def send_delete(self, stmt):
         delete = MySQLxCrud.Delete(

lib/mysqlx/result.py

@@ -448,18 +448,25 @@ def __init__(self, connection):
             connection._active_result = None
 
     @property
-    def Warnings(self):
+    def get_warnings(self):
         return self._warnings
 
+    def get_warnings_count(self):
+        return len(self._warnings)
+
 class Result(BaseResult):
     def __init__(self, connection):
         super(Result, self).__init__(connection)
         self._protocol.close_result(self)
 
     @property
-    def rows_affected(self):
+    def get_affected_items_count(self):
         return self._rows_affected
 
+    @property
+    def get_autoincrement_value(self):
+        pass
+
 class BufferingResult(BaseResult):
     def __init__(self, connection):
         super(BufferingResult, self).__init__(connection)

lib/mysqlx/statement.py

@@ -22,7 +22,7 @@
 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
 
 
-
+from .protobuf import mysqlx_crud_pb2 as MySQLxCrud
 from .result import SqlResult
 from .expr import ExprParser
 from .dbdoc import DbDoc
@@ -121,6 +121,52 @@ def execute(self):
             doc.ensure_id()
         return self._connection.send_insert(self)
 
+class UpdateSpec(object):
+    def __init__(self, type, source, value=None):
+        if type == MySQLxCrud.UpdateOperation.SET:
+            self._table_set(source, value)
+        else:
+            self.update_type = type
+            self.source = source
+            if len(source) > 0 and source[0] == '$':
+                self.source = source[1:]
+            self.source = ExprParser(self.source, False).document_field().identifier
+            self.value = value
+
+    def _table_set(self, source, value):
+        self.update_type = MySQLxCrud.UpdateOperation.SET
+        self.source = ExprParser(source, True).parse_table_update_field()
+        self.value = value
+
+class ModifyStatement(FilterableStatement):
+    def __init__(self, collection, condition=None):
+        super(ModifyStatement, self).__init__(target=collection, condition=condition)
+        self._update_ops = []
+
+    def set(self, doc_path, value):
+        self._update_ops.append(UpdateSpec(MySQLxCrud.UpdateOperation.ITEM_SET, doc_path, value))
+        return self
+
+    def change(self, doc_path, value):
+        self._update_ops.append(UpdateSpec(MySQLxCrud.UpdateOperation.ITEM_REPLACE, doc_path, value))
+        return self
+
+    def unset(self, doc_path):
+        self._update_ops.append(UpdateSpec(MySQLxCrud.UpdateOperation.ITEM_REMOVE, doc_path))
+        return self
+
+    def array_insert(self, field, value):
+        self._update_ops.append(UpdateSpec(MySQLxCrud.UpdateOperation.ARRAY_INSERT, field, value))
+        return self
+
+    def array_append(self, doc_path, value):
+        self._update_ops.append(UpdateSpec(MySQLxCrud.UpdateOperation.UpdateType.ARRAY_APPEND, doc_path, value))
+        return self
+
+    def execute(self):
+        return self._connection.update(self)
+
+
 class FindStatement(FilterableStatement):
     def __init__(self, collection, condition=None):
         super(FindStatement, self).__init__(collection, True, condition)
@@ -170,6 +216,19 @@ def values(self, *values):
     def execute(self):
         return self._connection.send_insert(self)
 
+class UpdateStatement(FilterableStatement):
+    def __init__(self, table, *fields):
+        super(UpdateStatement, self).__init__(target=table, doc_based=False)
+        self._update_ops = []
+
+    def set(self, field, value):
+        self._update_ops.append(UpdateSpec(MySQLxCrud.UpdateOperation.SET, field, value))
+        return self
+
+    def execute(self):
+        return self._connection.update(self)
+
+
 class RemoveStatement(FilterableStatement):
     def __init__(self, collection):
         super(RemoveStatement, self).__init__(target=collection)
@@ -178,9 +237,9 @@ def execute(self):
         return self._connection.delete(self)
 
 
-class TableDeleteStatement(FilterableStatement):
+class DeleteStatement(FilterableStatement):
     def __init__(self, table, condition=None):
-        super(TableDeleteStatement, self).__init__(target=table,
+        super(DeleteStatement, self).__init__(target=table,
                                                    condition=condition,
                                                    doc_based=False)
 

tests/test_mysqlx_crud.py

@@ -181,17 +181,17 @@ def test_add(self):
         collection_name = "collection_test"
         collection = self.schema.create_collection(collection_name)
         result = collection.add({"name":"Fred", "age":21}).execute()
-        self.assertEqual(result.rows_affected, 1)
+        self.assertEqual(result.get_affected_items_count, 1)
         self.assertEqual(1, collection.count())
 
         # now add multiple dictionaries at once
         result = collection.add({"name": "Wilma", "age": 33}, {"name": "Barney", "age": 42}).execute()
-        self.assertEqual(result.rows_affected, 2)
+        self.assertEqual(result.get_affected_items_count, 2)
         self.assertEqual(3, collection.count())
 
         # now let's try adding strings
         result = collection.add('{"name": "Bambam", "age": 8}', '{"name": "Pebbles", "age": 8}').execute()
-        self.assertEqual(result.rows_affected, 2)
+        self.assertEqual(result.get_affected_items_count, 2)
         self.assertEqual(5, collection.count())
 
     def test_remove(self):
@@ -200,7 +200,7 @@ def test_remove(self):
         collection.add({"name":"Fred", "age":21}).execute()
         self.assertEqual(1, collection.count())
         result = collection.remove("age == 21").execute()
-        self.assertEqual(1, result.rows_affected)
+        self.assertEqual(1, result.get_affected_items_count)
         self.assertEqual(0, collection.count())
 
     def test_find(self):
@@ -229,6 +229,31 @@ def test_find(self):
         self.assertEqual(42, docs[1]["age"])
         self.assertEqual(1, len(docs[1].keys()))
 
+    def test_modify(self):
+        collection_name = "collection_test"
+        collection = self.schema.create_collection(collection_name)
+        result = collection.add(
+            {"name":"Fred", "age":21},
+            {"name": "Barney", "age": 28},
+            {"name": "Wilma", "age": 42},
+            {"name": "Betty", "age": 67},
+
+        ).execute()
+
+        result = collection.modify("age < 67").set("young", True).execute()
+        self.assertEqual(3, result.get_affected_items_count)
+        doc = collection.find("name = 'Fred'").execute().fetch_all()[0]
+        self.assertEqual(True, doc.young)
+
+        result = collection.modify("age == 28").change("young", False).execute()
+        self.assertEqual(1, result.get_affected_items_count)
+        docs = collection.find("young = True").execute().fetch_all()
+        self.assertEqual(2, len(docs))
+
+        result = collection.modify("young == True").unset("young").execute()
+        self.assertEqual(2, result.get_affected_items_count)
+        docs = collection.find("young = True").execute().fetch_all()
+        self.assertEqual(0, len(docs))
 
     def test_results(self):
         collection_name = "collection_test"
@@ -376,8 +401,18 @@ def test_insert(self):
         self.assertEqual(4, len(rows))
 
     def test_update(self):
-        # TODO: To implement
-        pass
+        table = self.schema.get_table("test")
+
+        self.node_session.sql("CREATE TABLE {0}.test(age INT, name VARCHAR(50), gender CHAR(1))".format(self.schema_name)).execute()
+
+        result = table.insert("age", "name") \
+            .values(21, 'Fred') \
+            .values(28, 'Barney') \
+            .values(42, 'Wilma') \
+            .values(67, 'Betty').execute()
+
+        result = table.update().set("age", 25).where("age == 21").execute()
+        self.assertEqual(1, result.get_affected_items_count)
 
     def test_delete(self):
         table_name = "table_test"