implemented table insert and cleaned up protcol code so collection and table insert are both using same protocol code

Author: Reggie Burnett

lib/mysqlx/connection.py

@@ -92,10 +92,8 @@ def _authenticate(self):
     def send_sql(self, sql, *args):
         self.protocol.send_execute_statement("sql", sql, args)
 
-    def send_doc_insert(self, statement):
-        self.protocol.send_doc_insert(statement.schema.name,
-                                  statement.target.name,
-                                  statement._docs)
+    def send_insert(self, statement):
+        self.protocol.send_insert(statement)
         return Result(self)
 
     def find(self, statement):

lib/mysqlx/crud.py

@@ -23,7 +23,7 @@
 
 from .statement import (AddStatement, RemoveStatement, TableDeleteStatement,
                         FindStatement, SelectStatement,
-                        CreateCollectionIndexStatement,
+                        CreateCollectionIndexStatement, InsertStatement,
                         DropCollectionIndexStatement)
 
 
@@ -179,7 +179,7 @@ def select(self, *fields):
         return SelectStatement(self, *fields)
 
     def insert(self, *fields):
-        pass
+        return InsertStatement(self, *fields)
 
     def update(self):
         pass

lib/mysqlx/protocol.py

@@ -164,18 +164,26 @@ def send_execute_statement(self, namespace, stmt, args):
         self._writer.write_message(MySQLx.ClientMessages.SQL_STMT_EXECUTE,
                                    stmt)
 
-    def send_doc_insert(self, schema, target, docs):
+    def send_insert(self, statement):
         insert = MySQLxCrud.Insert(
-            collection=MySQLxCrud.Collection(schema=schema, name=target),
-            data_model=MySQLxCrud.DOCUMENT)
-        for doc in docs:
+            data_model=MySQLxCrud.DOCUMENT if statement._doc_based else MySQLxCrud.TABLE,
+            collection=MySQLxCrud.Collection(name=statement.target.name, schema=statement.schema.name))
+        if hasattr(statement, '_fields'):
+            for field in statement._fields:
+                insert.projection.extend([ExprParser(field, not statement._doc_based).parse_table_insert_field()])
+        for value in statement._values:
             row = MySQLxCrud.Insert.TypedRow()
-            o = self.arg_object_to_expr(doc, False)
-            row.field.extend([o])
+            if isinstance(value, list):
+                for v in value:
+                    o = self.arg_object_to_expr(v, not statement._doc_based)
+                    row.field.extend([o])
+            else:
+                o = self.arg_object_to_expr(value, not statement._doc_based)
+                row.field.extend([o])
             insert.row.extend([row])
-
         self._writer.write_message(MySQLx.ClientMessages.CRUD_INSERT, insert)
 
+
     def _create_any(self, arg):
         if isinstance(arg, (str, unicode,)):
             val = MySQLxDatatypes.Scalar.String(value=arg)
@@ -253,21 +261,21 @@ def get_column_metadata(self, rs):
     def arg_object_to_expr(self, value, allow_relational):
         if value == None:
             return Expr.build_null_scalar()
-        value_type = type(value)
-        if value_type == type(True):
-            return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=ExprParser().build_bool_scalar(value))
+        #value_type = type(value)
+        if isinstance(value, bool):
+            return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=build_bool_scalar(value))
         elif isinstance(value, (int, long)):
-            return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=ExprParser().build_int_scalar(value))
+            return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=build_int_scalar(value))
         elif isinstance(value, (float)):
-            return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=ExprParser().build_double_scalar(value))
-        elif value_type == type(str):
+            return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=build_double_scalar(value))
+        elif isinstance(value, basestring):
             try:
                 expression = ExprParser(value, allow_relational).expr()
                 if expression.has_identifier():
-                    return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=ExprParser().build_string_scalar(value))
+                    return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=build_string_scalar(value))
                 return expression
             except Exception as e:
-                return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=ExprParser().build_string_scalar(value))
+                return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=build_string_scalar(value))
         elif isinstance(value, DbDoc):
             return MySQLxExpr.Expr(type=MySQLxExpr.Expr.LITERAL, literal=build_string_scalar(str(value)))
-        raise Exception("Unsupported type")
+        raise Exception("Unsupported type: " + str(type(value)))

lib/mysqlx/statement.py

@@ -106,20 +106,20 @@ def execute(self):
 class AddStatement(Statement):
     def __init__(self, collection):
         super(AddStatement, self).__init__(target=collection)
-        self._docs = []
+        self._values = []
 
     def add(self, *values):
         for val in values:
             if isinstance(val, DbDoc):
-                self._docs.append(val)
+                self._values.append(val)
             else:
-                self._docs.append(DbDoc(val))
+                self._values.append(DbDoc(val))
         return self
 
     def execute(self):
-        for doc in self._docs:
+        for doc in self._values:
             doc.ensure_id()
-        return self._connection.send_doc_insert(self)
+        return self._connection.send_insert(self)
 
 class FindStatement(FilterableStatement):
     def __init__(self, collection, condition=None):
@@ -157,6 +157,19 @@ def execute(self):
         return self._connection.find(self)
 
 
+class InsertStatement(Statement):
+    def __init__(self, table, *fields):
+        super(InsertStatement, self).__init__(target=table, doc_based=False)
+        self._fields = fields
+        self._values = []
+
+    def values(self, *values):
+        self._values.append(list(values))
+        return self
+
+    def execute(self):
+        return self._connection.send_insert(self)
+
 class RemoveStatement(FilterableStatement):
     def __init__(self, collection):
         super(RemoveStatement, self).__init__(target=collection)

tests/test_mysqlx_crud.py

@@ -361,8 +361,19 @@ def test_having(self):
         self.assertEqual(42, rows[0]["age"])
 
     def test_insert(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.select().execute()
+        rows = result.fetch_all()
+        self.assertEqual(4, len(rows))
 
     def test_update(self):
         # TODO: To implement