api: extend connect with fetch_schema param

Added support of the fetch_schema parameter,
which allows to ignore schema changes on the server.

Closes #219

Author: GRISHNOV

.gitignore

@@ -53,3 +53,5 @@ test/data/*.key
 !test/data/localhost.enc.key
 test/data/*.pem
 test/data/*.srl
+
+.rocks

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   to decrypt private SSL key file (#224).
 - Support specifying authentication method with `auth_type`
   and Tarantool EE `pap-sha256` authentication method (#269).
+- Support of the fetch_schema parameter for connection (#219).
 
 ## 0.10.0 - 2022-11-09
 

docs/source/quick-start.rst

@@ -359,7 +359,7 @@ Through the :class:`~tarantool.Connection` object, you can access
 
     >>> import tarantool
     >>> from tarantool.error import CrudModuleError, CrudModuleManyError, DatabaseError
-    >>> conn = tarantool.Connection(host='localhost',port=3301)
+    >>> conn = tarantool.Connection(host='localhost',port=3301,fetch_schema=False)
 
     >>> conn.crud_
     conn.crud_count(                conn.crud_insert(               conn.crud_insert_object_many(   

tarantool/connection.py

@@ -579,7 +579,8 @@ def __init__(self, host, port,
                  ssl_password_file=DEFAULT_SSL_PASSWORD_FILE,
                  packer_factory=default_packer_factory,
                  unpacker_factory=default_unpacker_factory,
-                 auth_type=None):
+                 auth_type=None,
+                 fetch_schema=True):
         """
         :param host: Server hostname or IP address. Use ``None`` for
             Unix sockets.
@@ -736,6 +737,21 @@ def __init__(self, host, port,
             ``"chap-sha1"``.
         :type auth_type: :obj:`None` or :obj:`str`, optional
 
+        :param bool fetch_schema: If ``False``, allows to ignore schema
+            changes on the server, schema updates are not automatically
+            loaded. As a result, these methods become unavailable:
+            :meth:`~tarantool.Connection.replace`,
+            :meth:`~tarantool.Connection.insert`,
+            :meth:`~tarantool.Connection.delete`,
+            :meth:`~tarantool.Connection.upsert`,
+            :meth:`~tarantool.Connection.update`,
+            :meth:`~tarantool.Connection.select`,
+            :meth:`~tarantool.Connection.space`.
+            Calling the :meth:`~tarantool.Connection.update_schema`
+            method causes this parameter to switch
+            from ``False`` to ``True``.
+        :type fetch_schema: :obj:`bool`, optional
+
         :raise: :exc:`~tarantool.error.ConfigurationError`,
             :meth:`~tarantool.Connection.connect` exceptions
 
@@ -766,8 +782,9 @@ def __init__(self, host, port,
         self.socket_timeout = socket_timeout
         self.reconnect_delay = reconnect_delay
         self.reconnect_max_attempts = reconnect_max_attempts
+        self.fetch_schema = fetch_schema
         self.schema = Schema(self)
-        self.schema_version = 1
+        self.schema_version = 0
         self._socket = None
         self.connected = False
         self.error = True
@@ -1023,7 +1040,8 @@ def connect(self):
             if self.transport == SSL_TRANSPORT:
                 self.wrap_socket_ssl()
             self.handshake()
-            self.load_schema()
+            if self.schema.schema != {}:
+                self.load_schema()
         except SslError as e:
             raise e
         except Exception as e:
@@ -1118,7 +1136,8 @@ def _send_request_wo_reconnect(self, request, on_push=None, on_push_ctx=None):
                 response = request.response_class(self, self._read_response())
                 break
             except SchemaReloadException as e:
-                self.update_schema(e.schema_version)
+                if self.schema.schema != {}:
+                    self.update_schema(e.schema_version)
                 continue
 
         while response._code == IPROTO_CHUNK:
@@ -1255,6 +1274,9 @@ def update_schema(self, schema_version):
         :meta private:
         """
 
+        if self.fetch_schema == False:
+            self.fetch_schema = True
+
         self.schema_version = schema_version
         self.flush_schema()
 
@@ -1269,6 +1291,18 @@ def flush_schema(self):
         self.schema.flush()
         self.load_schema()
 
+    def _fetch_schema_support_check(self):
+        """
+        Checks the fetch_schema opt of the connection is enabled.
+        If the opt is disabled, an exception will be thrown 
+        about unsupporting the method with fetch_schema=False.
+
+        :raise: :exc:`~tarantool.error.NotSupportedError`
+        """
+        if not self.fetch_schema:
+            raise NotSupportedError('This method is not available in ' +
+                                    'connection opened with fetch_schema=False')
+
     def call(self, func_name, *args, on_push=None, on_push_ctx=None):
         """
         Execute a CALL request: call a stored Lua function.
@@ -1366,11 +1400,14 @@ def replace(self, space_name, values, on_push=None, on_push_ctx=None):
             :exc:`~tarantool.error.DatabaseError`,
             :exc:`~tarantool.error.SchemaError`,
             :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
+            :exc:`~tarantool.error.SslError`,
+            :exc:`~tarantool.error.NotSupportedError`
 
         .. _replace: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/replace/
         """
 
+        self._fetch_schema_support_check()
+
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         if on_push is not None and not callable(on_push):
@@ -1411,7 +1448,7 @@ def authenticate(self, user, password):
                               password=self.password,
                               auth_type=self._get_auth_type())
         auth_response = self._send_request_wo_reconnect(request)
-        if auth_response.return_code == 0:
+        if auth_response.return_code == 0 and self.schema.schema != {}:
             self.flush_schema()
         return auth_response
 
@@ -1584,11 +1621,14 @@ def insert(self, space_name, values, on_push=None, on_push_ctx=None):
             :exc:`~tarantool.error.DatabaseError`,
             :exc:`~tarantool.error.SchemaError`,
             :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
+            :exc:`~tarantool.error.SslError`,
+            :exc:`~tarantool.error.NotSupportedError`
 
         .. _insert: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/insert/
         """
 
+        self._fetch_schema_support_check()
+
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         if on_push is not None and not callable(on_push):
@@ -1623,11 +1663,14 @@ def delete(self, space_name, key, *, index=0, on_push=None, on_push_ctx=None):
             :exc:`~tarantool.error.DatabaseError`,
             :exc:`~tarantool.error.SchemaError`,
             :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
+            :exc:`~tarantool.error.SslError`,
+            :exc:`~tarantool.error.NotSupportedError`
 
         .. _delete: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/delete/
         """
 
+        self._fetch_schema_support_check()
+
         key = wrap_key(key)
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
@@ -1682,11 +1725,14 @@ def upsert(self, space_name, tuple_value, op_list, *, index=0, on_push=None, on_
             :exc:`~tarantool.error.DatabaseError`,
             :exc:`~tarantool.error.SchemaError`,
             :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
+            :exc:`~tarantool.error.SslError`,
+            :exc:`~tarantool.error.NotSupportedError`
 
         .. _upsert: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/upsert/
         """
 
+        self._fetch_schema_support_check()
+
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         if isinstance(index, str):
@@ -1770,11 +1816,14 @@ def update(self, space_name, key, op_list, *, index=0, on_push=None, on_push_ctx
             :exc:`~tarantool.error.DatabaseError`,
             :exc:`~tarantool.error.SchemaError`,
             :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
+            :exc:`~tarantool.error.SslError`,
+            :exc:`~tarantool.error.NotSupportedError`
 
         .. _update: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/update/
         """
 
+        self._fetch_schema_support_check()
+
         key = wrap_key(key)
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
@@ -1956,11 +2005,14 @@ def select(self, space_name, key=None, *, offset=0, limit=0xffffffff, index=0, i
             :exc:`~tarantool.error.DatabaseError`,
             :exc:`~tarantool.error.SchemaError`,
             :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
+            :exc:`~tarantool.error.SslError`,
+            :exc:`~tarantool.error.NotSupportedError`
 
         .. _select: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/select/
         """
 
+        self._fetch_schema_support_check()
+
         if iterator is None:
             iterator = ITERATOR_EQ
             if key is None or (isinstance(key, (list, tuple)) and
@@ -1996,6 +2048,8 @@ def space(self, space_name):
         :raise: :exc:`~tarantool.error.SchemaError`
         """
 
+        self._fetch_schema_support_check()
+
         return Space(self, space_name)
 
     def generate_sync(self):

tarantool/connection_pool.py

@@ -378,7 +378,8 @@ def __init__(self,
                  call_16=False,
                  connection_timeout=CONNECTION_TIMEOUT,
                  strategy_class=RoundRobinStrategy,
-                 refresh_delay=POOL_REFRESH_DELAY):
+                 refresh_delay=POOL_REFRESH_DELAY,
+                 fetch_schema=True):
         """
         :param addrs: List of dictionaries describing server addresses:
 
@@ -452,6 +453,9 @@ def __init__(self,
             `box.info.ro`_ status background refreshes, in seconds.
         :type connection_timeout: :obj:`float`, optional
 
+        :param fetch_schema: Refer to
+            :paramref:`~tarantool.Connection.params.fetch_schema`.
+
         :raise: :exc:`~tarantool.error.ConfigurationError`,
             :class:`~tarantool.Connection` exceptions
 
@@ -500,7 +504,8 @@ def __init__(self,
                     ssl_ciphers=addr['ssl_ciphers'],
                     ssl_password=addr['ssl_password'],
                     ssl_password_file=addr['ssl_password_file'],
-                    auth_type=addr['auth_type'])
+                    auth_type=addr['auth_type'],
+                    fetch_schema=fetch_schema)
             )
 
         if connect_now:

tarantool/mesh_connection.py

@@ -283,7 +283,8 @@ def __init__(self, host=None, port=None,
                  addrs=None,
                  strategy_class=RoundRobinStrategy,
                  cluster_discovery_function=None,
-                 cluster_discovery_delay=CLUSTER_DISCOVERY_DELAY):
+                 cluster_discovery_delay=CLUSTER_DISCOVERY_DELAY,
+                 fetch_schema=True):
         """
         :param host: Refer to
             :paramref:`~tarantool.Connection.params.host`.
@@ -425,6 +426,9 @@ def __init__(self, host=None, port=None,
             list refresh.
         :type cluster_discovery_delay: :obj:`float`, optional
 
+        :param fetch_schema: Refer to
+            :paramref:`~tarantool.Connection.params.fetch_schema`.
+
         :raises: :exc:`~tarantool.error.ConfigurationError`,
             :class:`~tarantool.Connection` exceptions,
             :class:`~tarantool.MeshConnection.connect` exceptions
@@ -489,7 +493,8 @@ def __init__(self, host=None, port=None,
             ssl_ciphers=addr['ssl_ciphers'],
             ssl_password=addr['ssl_password'],
             ssl_password_file=addr['ssl_password_file'],
-            auth_type=addr['auth_type'])
+            auth_type=addr['auth_type'],
+            fetch_schema=fetch_schema)
 
     def connect(self):
         """

tarantool/request.py

@@ -188,9 +188,13 @@ def header(self, length):
         """
 
         self._sync = self.conn.generate_sync()
-        header = self._dumps({IPROTO_REQUEST_TYPE: self.request_type,
-                              IPROTO_SYNC: self._sync,
-                              IPROTO_SCHEMA_ID: self.conn.schema_version})
+        header_fields = {
+            IPROTO_REQUEST_TYPE: self.request_type,
+            IPROTO_SYNC: self._sync,
+        }
+        if self.conn.schema.schema != {}:
+            header_fields[IPROTO_SCHEMA_ID] = self.conn.schema_version
+        header = self._dumps(header_fields)
 
         return self._dumps(length + len(header)) + header
 

test/suites/crud_server.lua

@@ -71,17 +71,15 @@ box.schema.user.grant(
 )
 
 if crud_imported == false or vshard_imported == false then
+    -- Set flag for unittest.
+    _G['ROCKS_IMPORT_FAIL'] = true
     local fail_msg = 'The crud/vshard modules are not detected, ' ..
                      'installation via rocks install is required ' ..
                      'for CRUD testing purposes. You can use ' ..
                      '<tarantoolctl rocks install crud> or ' ..
                      '<tt rocks install crud> to install modules'
     -- The print output will be captured in the logs.
     print(fail_msg)
-    -- Because of the implementation of the testing framework,
-    -- this will cause the crash of the CRUD test and display to user
-    -- a message about require fail during instance configuration.
-    box.info.status = fail_msg
 else
     configure_crud_instance(primary_listen, crud, vshard)
 end

test/suites/test_crud.py

@@ -33,13 +33,20 @@ def setUp(self):
         time.sleep(1)
         # Open connections to instance.
         self.conn = tarantool.Connection(host=self.host, port=self.port, 
-                                         user='guest', password='')
+                                         user='guest', password='', fetch_schema=False)
         self.conn_mesh = tarantool.MeshConnection(host=self.host, port=self.port, 
-                                         user='guest', password='')
+                                         user='guest', password='', fetch_schema=False)
         self.conn_pool = tarantool.ConnectionPool([{'host':self.host, 'port':self.port}], 
-                                                     user='guest', password='')
+                                                     user='guest', password='', 
+                                                     fetch_schema=False)
         # Time for vshard group configuration.
         time.sleep(1)
+        if self.conn.eval('return ROCKS_IMPORT_FAIL').data[0] == True:
+            raise unittest.SkipTest('The crud/vshard modules are not detected, ' +
+                                    'installation via rocks install is required ' +
+                                    'for CRUD testing purposes. You can use ' +
+                                    '<tarantoolctl rocks install crud> or ' +
+                                    '<tt rocks install crud> to install modules')
 
     crud_test_cases = {
         'crud_insert': {