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

@@ -41,3 +41,5 @@ debian/files
 debian/*.substvars
 
 deb_dist
+
+.rocks

CHANGELOG.md

@@ -8,8 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 - Support custom packer and unpacker factories (#191).
-
 - Support [crud module](https://github.com/tarantool/crud) native API (#205).
+- Support of the fetch_schema parameter for connection (#219).
 
 ### Changed
 

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

@@ -570,7 +570,8 @@ def __init__(self, host, port,
                  ssl_ca_file=DEFAULT_SSL_CA_FILE,
                  ssl_ciphers=DEFAULT_SSL_CIPHERS,
                  packer_factory=default_packer_factory,
-                 unpacker_factory=default_unpacker_factory):
+                 unpacker_factory=default_unpacker_factory,
+                 fetch_schema=True):
         """
         :param host: Server hostname or IP address. Use ``None`` for
             Unix sockets.
@@ -710,6 +711,18 @@ def __init__(self, host, port,
             callable[[:obj:`~tarantool.Connection`], :obj:`~msgpack.Unpacker`],
             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`.
+        :type fetch_schema: :obj:`bool`, optional
+
         :raise: :exc:`~tarantool.error.ConfigurationError`,
             :meth:`~tarantool.Connection.connect` exceptions
 
@@ -740,6 +753,7 @@ 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._socket = None
@@ -952,7 +966,8 @@ def connect(self):
             if self.transport == SSL_TRANSPORT:
                 self.wrap_socket_ssl()
             self.handshake()
-            self.load_schema()
+            if self.fetch_schema:
+                self.load_schema()
             self._check_features()
         except SslError as e:
             raise e
@@ -1048,7 +1063,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.fetch_schema:
+                    self.update_schema(e.schema_version)
                 continue
 
         while response._code == IPROTO_CHUNK:
@@ -1199,6 +1215,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.
@@ -1296,11 +1324,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):
@@ -1338,7 +1369,7 @@ def authenticate(self, user, password):
         request = RequestAuthenticate(self, self._salt, self.user,
                                       self.password)
         auth_response = self._send_request_wo_reconnect(request)
-        if auth_response.return_code == 0:
+        if auth_response.return_code == 0 and self.fetch_schema:
             self.flush_schema()
         return auth_response
 
@@ -1483,11 +1514,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):
@@ -1522,11 +1556,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
@@ -1581,11 +1618,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):
@@ -1669,11 +1709,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
@@ -1855,11 +1898,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
@@ -1895,6 +1941,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

@@ -376,7 +376,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:
 
@@ -447,6 +448,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
 
@@ -492,7 +496,8 @@ def __init__(self,
                     ssl_key_file=addr['ssl_key_file'],
                     ssl_cert_file=addr['ssl_cert_file'],
                     ssl_ca_file=addr['ssl_ca_file'],
-                    ssl_ciphers=addr['ssl_ciphers'])
+                    ssl_ciphers=addr['ssl_ciphers'],
+                    fetch_schema=fetch_schema)
             )
 
         if connect_now:

tarantool/mesh_connection.py

@@ -272,7 +272,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`.
@@ -409,6 +410,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
@@ -467,7 +471,8 @@ def __init__(self, host=None, port=None,
             ssl_key_file=addr['ssl_key_file'],
             ssl_cert_file=addr['ssl_cert_file'],
             ssl_ca_file=addr['ssl_ca_file'],
-            ssl_ciphers=addr['ssl_ciphers'])
+            ssl_ciphers=addr['ssl_ciphers'],
+            fetch_schema=fetch_schema)
 
     def connect(self):
         """

tarantool/request.py

@@ -186,9 +186,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.fetch_schema:
+            header_fields[IPROTO_SCHEMA_ID] = self.conn.schema_version
+        header = self._dumps(header_fields)
 
         return self._dumps(length + len(header)) + header