Add sql execute to connection

Closes #159

Co-authored-by Denis Ignatenko <denis.ignatenko@gamil.com>

Author: artembo

tarantool/connection.py

@@ -34,7 +34,8 @@
     RequestSubscribe,
     RequestUpdate,
     RequestUpsert,
-    RequestAuthenticate
+    RequestAuthenticate,
+    RequestExecute
 )
 from tarantool.space import Space
 from tarantool.const import (
@@ -250,17 +251,18 @@ def _read_response(self):
 
     def _send_request_wo_reconnect(self, request):
         '''
-        :rtype: `Response` instance
+        :rtype: `Response` instance or subclass
 
         :raise: NetworkError
         '''
-        assert isinstance(request, Request)
+        if not issubclass(type(request), Request):
+            raise NetworkError
 
         response = None
         while True:
             try:
                 self._socket.sendall(bytes(request))
-                response = Response(self, self._read_response())
+                response = request.response_class(self, self._read_response())
                 break
             except SchemaReloadException as e:
                 self.update_schema(e.schema_version)
@@ -785,3 +787,22 @@ def generate_sync(self):
         Need override for async io connection
         '''
         return 0
+
+    def execute(self, query, params=None):
+        '''
+        Execute SQL request.
+
+        :param query: SQL syntax query
+        :type query: str
+
+        :param params: Bind values to use in query
+        :type params: list, dict
+
+        :return: query result data
+        :rtype: list
+        '''
+        if not params:
+            params = []
+        request = RequestExecute(self, query, params)
+        response = self._send_request(request)
+        return response

tarantool/const.py

@@ -29,6 +29,13 @@
 #
 IPROTO_DATA = 0x30
 IPROTO_ERROR = 0x31
+#
+IPROTO_METADATA = 0x32
+IPROTO_SQL_TEXT = 0x40
+IPROTO_SQL_BIND = 0x41
+IPROTO_SQL_INFO = 0x42
+IPROTO_SQL_INFO_ROW_COUNT = 0x00
+IPROTO_SQL_INFO_AUTOINCREMENT_IDS = 0x01
 
 IPROTO_GREETING_SIZE = 128
 IPROTO_BODY_MAX_LEN = 2147483648
@@ -44,6 +51,7 @@
 REQUEST_TYPE_EVAL = 8
 REQUEST_TYPE_UPSERT = 9
 REQUEST_TYPE_CALL = 10
+REQUEST_TYPE_EXECUTE = 11
 REQUEST_TYPE_PING = 64
 REQUEST_TYPE_JOIN = 65
 REQUEST_TYPE_SUBSCRIBE = 66

tarantool/request.py

@@ -7,7 +7,7 @@
 import msgpack
 import hashlib
 
-
+from tarantool.error import DatabaseError
 from tarantool.const import (
     IPROTO_CODE,
     IPROTO_SYNC,
@@ -27,6 +27,8 @@
     IPROTO_OPS,
     # IPROTO_INDEX_BASE,
     IPROTO_SCHEMA_ID,
+    IPROTO_SQL_TEXT,
+    IPROTO_SQL_BIND,
     REQUEST_TYPE_OK,
     REQUEST_TYPE_PING,
     REQUEST_TYPE_SELECT,
@@ -37,11 +39,13 @@
     REQUEST_TYPE_UPSERT,
     REQUEST_TYPE_CALL16,
     REQUEST_TYPE_CALL,
+    REQUEST_TYPE_EXECUTE,
     REQUEST_TYPE_EVAL,
     REQUEST_TYPE_AUTHENTICATE,
     REQUEST_TYPE_JOIN,
     REQUEST_TYPE_SUBSCRIBE
 )
+from tarantool.response import Response, ResponseExecute
 from tarantool.utils import (
     strxor,
     binary_types
@@ -64,6 +68,7 @@ def __init__(self, conn):
         self.conn = conn
         self._sync = None
         self._body = ''
+        self.response_class = Response
 
     def __bytes__(self):
         return self.header(len(self._body)) + self._body
@@ -332,3 +337,24 @@ def __init__(self, conn, sync):
         request_body = msgpack.dumps({IPROTO_CODE: self.request_type,
                                       IPROTO_SYNC: sync})
         self._body = request_body
+
+
+class RequestExecute(Request):
+    '''
+    Represents EXECUTE request
+    '''
+    request_type = REQUEST_TYPE_EXECUTE
+
+    # pylint: disable=W0231
+    def __init__(self, conn, sql, args):
+        super(RequestExecute, self).__init__(conn)
+        if isinstance(args, dict):
+            args = [{":%s" % name: value} for name, value in args.items()]
+        try:
+            request_body = msgpack.dumps({IPROTO_SQL_TEXT: sql,
+                                          IPROTO_SQL_BIND: args})
+        except ValueError as e:
+            raise DatabaseError("Value error: %s" % e)
+
+        self._body = request_body
+        self.response_class = ResponseExecute

tarantool/response.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 # pylint: disable=C0301,W0105,W0401,W0614
+from pprint import pprint
 
 try:
     # Python 3.3+
@@ -17,7 +18,10 @@
     IPROTO_ERROR,
     IPROTO_SYNC,
     IPROTO_SCHEMA_ID,
-    REQUEST_TYPE_ERROR
+    REQUEST_TYPE_ERROR,
+    IPROTO_SQL_INFO,
+    IPROTO_SQL_INFO_ROW_COUNT,
+    IPROTO_SQL_INFO_AUTOINCREMENT_IDS
 )
 from tarantool.error import (
     DatabaseError,
@@ -248,3 +252,34 @@ def __str__(self):
         return ''.join(output)
 
     __repr__ = __str__
+
+
+class ResponseExecute(Response):
+    @property
+    def lastrowid(self):
+        if self.body is None:
+            raise InterfaceError("Trying to access data, when there's no data")
+        info = self.body.get(IPROTO_SQL_INFO)
+
+        if info is None:
+            return None
+
+        lastrowids = info.get(IPROTO_SQL_INFO_AUTOINCREMENT_IDS)
+
+        return lastrowids[-1] if lastrowids else None
+
+    @property
+    def rowcount(self):
+        if self._body is None:
+            raise InterfaceError("Trying to access data, when there's no data")
+
+        info = self._body.get(IPROTO_SQL_INFO)
+
+        if info is None:
+            return -1
+
+        return info.get(IPROTO_SQL_INFO_ROW_COUNT, -1)
+
+    @property
+    def rows(self):
+        return self._data if self._body.get(IPROTO_DATA) is not None else None