Merge pull request #651 from dimitri-yatsenko/refactor-errors

Refactor error handling and fix schema diagrams

Author: guzman-raphael

CHANGELOG.md

@@ -24,6 +24,9 @@
 ### 0.11.3 -- Jul 26, 2019
 * Fix incompatibility with pyparsing 2.4.1 (#629) PR #631
 
+### 0.11.2 -- July 25, 2019
+* Fix #628 - incompatibility with pyparsing 2.4.1
+
 ### 0.11.1 -- Nov 15, 2018
 * Fix ordering of attributes in proj (#483 and #516)
 * Prohibit direct insert into auto-populated tables (#511)

datajoint/__init__.py

@@ -24,8 +24,7 @@
            'Not', 'AndList', 'U', 'Diagram', 'Di', 'ERD',
            'set_password', 'kill',
            'MatCell', 'MatStruct',
-           'DataJointError', 'DuplicateError', 'key']
-
+           'errors', 'DataJointError', 'key']
 
 from .version import __version__
 from .settings import config
@@ -38,7 +37,8 @@
 from .diagram import Diagram
 from .admin import set_password, kill
 from .blob import MatCell, MatStruct
-from .errors import DataJointError, DuplicateError
 from .fetch import key
+from . import errors
+from .errors import DataJointError
 
 ERD = Di = Diagram   # Aliases for Diagram

datajoint/connection.py

@@ -7,12 +7,48 @@
 import pymysql as client
 import logging
 from getpass import getpass
-from pymysql import err
 
 from .settings import config
-from .errors import DataJointError, server_error_codes, is_connection_error
+from . import errors
 from .dependencies import Dependencies
 
+# client errors to catch
+client_errors = (client.err.InterfaceError, client.err.DatabaseError)
+
+
+def translate_query_error(client_error, query):
+    """
+    Take client error and original query and return the corresponding DataJoint exception.
+    :param client_error: the exception raised by the client interface
+    :param query: sql query with placeholders
+    :return: an instance of the corresponding subclass of datajoint.errors.DataJointError
+    """
+    # Loss of connection errors
+    if isinstance(client_error, client.err.InterfaceError) and client_error.args[0] == "(0, '')":
+        return errors.LostConnectionError('Server connection lost due to an interface error.', *client_error.args[1:])
+    disconnect_codes = {
+        2006: "Connection timed out",
+        2013: "Server connection lost"}
+    if isinstance(client_error, client.err.OperationalError) and client_error.args[0] in disconnect_codes:
+        return errors.LostConnectionError(disconnect_codes[client_error.args[0]], *client_error.args[1:])
+    # Access errors
+    if isinstance(client_error, client.err.OperationalError) and client_error.args[0] in (1044, 1142):
+        return errors.AccessError('Insufficient privileges.', client_error.args[1],  query)
+    # Integrity errors
+    if isinstance(client_error, client.err.IntegrityError) and client_error.args[0] == 1062:
+        return errors.DuplicateError(*client_error.args[1:])
+    if isinstance(client_error, client.err.IntegrityError) and client_error.args[0] == 1452:
+        return errors.IntegrityError(*client_error.args[1:])
+    # Syntax Errors
+    if isinstance(client_error, client.err.ProgrammingError) and client_error.args[0] == 1064:
+        return errors.QuerySyntaxError(client_error.args[1], query)
+    # Existence Errors
+    if isinstance(client_error, client.err.ProgrammingError) and client_error.args[0] == 1146:
+        return errors.MissingTableError(client_error.args[1], query)
+    if isinstance(client_error, client.err.InternalError) and client_error.args[0] == 1364:
+        return errors.MissingAttributeError(*client_error.args[1:])
+    raise client_error
+
 
 logger = logging.getLogger(__name__)
 
@@ -60,6 +96,7 @@ class Connection:
     :param init_fun: connection initialization function (SQL)
     :param use_tls: TLS encryption option
     """
+
     def __init__(self, host, user, password, port=None, init_fun=None, use_tls=None):
         if ':' in host:
             # the port in the hostname overrides the port argument
@@ -79,7 +116,7 @@ def __init__(self, host, user, password, port=None, init_fun=None, use_tls=None)
             logger.info("Connected {user}@{host}:{port}".format(**self.conn_info))
             self.connection_id = self.query('SELECT connection_id()').fetchone()[0]
         else:
-            raise DataJointError('Connection failed.')
+            raise errors.ConnectionError('Connection failed.')
         self._in_transaction = False
         self.schemas = dict()
         self.dependencies = Dependencies(self)
@@ -103,16 +140,16 @@ def connect(self):
                 self._conn = client.connect(
                     init_command=self.init_fun,
                     sql_mode="NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO,"
-                            "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION",
+                             "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION",
                     charset=config['connection.charset'],
                     **self.conn_info)
-            except err.InternalError:
+            except client.err.InternalError:
                 if ssl_input is None:
                     self.conn_info.pop('ssl')
                 self._conn = client.connect(
                     init_command=self.init_fun,
                     sql_mode="NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO,"
-                            "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION",
+                             "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION",
                     charset=config['connection.charset'],
                     **self.conn_info)
         self.conn_info['ssl_input'] = ssl_input
@@ -141,50 +178,46 @@ def is_connected(self):
             return False
         return True
 
-    def query(self, query, args=(), as_dict=False, suppress_warnings=True, reconnect=None):
+    @staticmethod
+    def __execute_query(cursor, query, args, cursor_class, suppress_warnings):
+        try:
+            with warnings.catch_warnings():
+                if suppress_warnings:
+                    # suppress all warnings arising from underlying SQL library
+                    warnings.simplefilter("ignore")
+                cursor.execute(query, args)
+        except client_errors as err:
+            raise translate_query_error(err, query)
+
+    def query(self, query, args=(), *, as_dict=False, suppress_warnings=True, reconnect=None):
         """
         Execute the specified query and return the tuple generator (cursor).
-
-        :param query: mysql query
+        :param query: SQL query
         :param args: additional arguments for the client.cursor
         :param as_dict: If as_dict is set to True, the returned cursor objects returns
                         query results as dictionary.
         :param suppress_warnings: If True, suppress all warnings arising from underlying query library
+        :param reconnect: when None, get from config, when True, attempt to reconnect if disconnected
         """
         if reconnect is None:
             reconnect = config['database.reconnect']
-
-        cursor = client.cursors.DictCursor if as_dict else client.cursors.Cursor
-        cur = self._conn.cursor(cursor=cursor)
-
         logger.debug("Executing SQL:" + query[0:300])
+        cursor_class = client.cursors.DictCursor if as_dict else client.cursors.Cursor
+        cursor = self._conn.cursor(cursor=cursor_class)
         try:
-            with warnings.catch_warnings():
-                if suppress_warnings:
-                    # suppress all warnings arising from underlying SQL library
-                    warnings.simplefilter("ignore")
-                cur.execute(query, args)
-        except (err.InterfaceError, err.OperationalError) as e:
-            if is_connection_error(e) and reconnect:
-                warnings.warn("Mysql server has gone away. Reconnecting to the server.")
-                self.connect()
-                if self._in_transaction:
-                    self.cancel_transaction()
-                    raise DataJointError("Connection was lost during a transaction.") from None
-                else:
-                    logger.debug("Re-executing SQL")
-                    cur = self.query(query, args=args, as_dict=as_dict, suppress_warnings=suppress_warnings, reconnect=False)
-            else:
-                logger.debug("Caught InterfaceError/OperationalError.")
+            self.__execute_query(cursor, query, args, cursor_class, suppress_warnings)
+        except errors.LostConnectionError:
+            if not reconnect:
                 raise
-        except err.ProgrammingError as e:
-            if e.args[0] == server_error_codes['parse error']:
-                raise DataJointError("\n".join((
-                    "Error in query:", query,
-                    "Please check spelling, syntax, and existence of tables and attributes.",
-                    "When restricting a relation by a condition in a string, enclose attributes in backquotes."
-                ))) from None
-        return cur
+            warnings.warn("MySQL server has gone away. Reconnecting to the server.")
+            self.connect()
+            if self._in_transaction:
+                self.cancel_transaction()
+                raise errors.LostConnectionError("Connection was lost during a transaction.") from None
+            logger.debug("Re-executing")
+            cursor = self._conn.cursor(cursor=cursor_class)
+            self.__execute_query(cursor, query, args, cursor_class, suppress_warnings)
+        return cursor
 
     def get_user(self):
         """
@@ -204,11 +237,9 @@ def in_transaction(self):
     def start_transaction(self):
         """
         Starts a transaction error.
-
-        :raise DataJointError: if there is an ongoing transaction.
         """
         if self.in_transaction:
-            raise DataJointError("Nested connections are not supported.")
+            raise errors.DataJointError("Nested connections are not supported.")
         self.query('START TRANSACTION WITH CONSISTENT SNAPSHOT')
         self._in_transaction = True
         logger.info("Transaction started")
@@ -252,3 +283,4 @@ def transaction(self):
             raise
         else:
             self.commit_transaction()
+            

datajoint/declare.py

@@ -235,7 +235,7 @@ def prepare_declare(definition, context):
     external_stores = []
 
     for line in definition:
-        if line.startswith('#'):  # ignore additional comments
+        if not line or line.startswith('#'):  # ignore additional comments
             pass
         elif line.startswith('---') or line.startswith('___'):
             in_key = False  # start parsing dependent attributes

datajoint/dependencies.py

@@ -62,10 +62,10 @@ def load(self):
         # add edges to the graph
         for fk in fks.values():
             props = dict(
-                primary=all(attr in pks[fk['referencing_table']] for attr in fk['attr_map']),
+                primary=set(fk['attr_map']) <= set(pks[fk['referencing_table']]),
                 attr_map=fk['attr_map'],
                 aliased=any(k != v for k, v in fk['attr_map'].items()),
-                multi=not all(a in fk['attr_map'] for a in pks[fk['referencing_table']]))
+                multi=set(fk['attr_map']) != set(pks[fk['referencing_table']]))
             if not props['aliased']:
                 self.add_edge(fk['referenced_table'], fk['referencing_table'], **props)
             else:

datajoint/diagram.py

@@ -191,6 +191,8 @@ def __add__(self, arg):
                         new = nx.algorithms.boundary.node_boundary(self, self.nodes_to_show)
                         if not new:
                             break
+                        # add nodes referenced by aliased nodes
+                        new.update(nx.algorithms.boundary.node_boundary(self, (a for a in new if a.isdigit())))
                         self.nodes_to_show.update(new)
             return self
 
@@ -207,9 +209,12 @@ def __sub__(self, arg):
                     self.nodes_to_show.remove(arg.full_table_name)
                 except AttributeError:
                     for i in range(arg):
-                        new = nx.algorithms.boundary.node_boundary(nx.DiGraph(self).reverse(), self.nodes_to_show)
+                        graph = nx.DiGraph(self).reverse()
+                        new = nx.algorithms.boundary.node_boundary(graph, self.nodes_to_show)
                         if not new:
                             break
+                        # add nodes referenced by aliased nodes
+                        new.update(nx.algorithms.boundary.node_boundary(graph, (a for a in new if a.isdigit())))
                         self.nodes_to_show.update(new)
             return self
 
@@ -229,9 +234,10 @@ def _make_graph(self):
             """
             # mark "distinguished" tables, i.e. those that introduce new primary key attributes
             for name in self.nodes_to_show:
-                foreign_attributes = set(attr for p in self.in_edges(name, data=True) for attr in p[2]['attr_map'])
-                self.node[name]['distinguished'] = ('primary_key' in self.node[name] and
-                        foreign_attributes < self.node[name]['primary_key'])
+                foreign_attributes = set(
+                    attr for p in self.in_edges(name, data=True) for attr in p[2]['attr_map'] if p[2]['primary'])
+                self.node[name]['distinguished'] = (
+                        'primary_key' in self.node[name] and foreign_attributes < self.node[name]['primary_key'])
             # include aliased nodes that are sandwiched between two displayed nodes
             gaps = set(nx.algorithms.boundary.node_boundary(self, self.nodes_to_show)).intersection(
                 nx.algorithms.boundary.node_boundary(nx.DiGraph(self).reverse(), self.nodes_to_show))

datajoint/errors.py

@@ -1,40 +1,76 @@
-from pymysql import err
+"""
+Exception classes for the DataJoint library
+"""
 
-server_error_codes = {
-    'unknown column': 1054,
-    'duplicate entry': 1062,
-    'parse error': 1064,
-    'command denied': 1142,
-    'table does not exist': 1146,
-    'syntax error': 1149,
-}
 
-operation_error_codes = {
-    'connection timedout': 2006,
-    'lost connection': 2013,
-}
+# --- Top Level ---
+class DataJointError(Exception):
+    """
+    Base class for errors specific to DataJoint internal operation.
+    """
+    def suggest(self, *args):
+        """
+        regenerate the exception with additional arguments
+        :param args: addition arguments
+        :return: a new exception of the same type with the additional arguments
+        """
+        return self.__class__(*(self.args + args))
 
 
-def is_connection_error(e):
+# --- Second Level ---
+class LostConnectionError(DataJointError):
     """
-    Checks if error e pertains to a connection issue
+    Loss of server connection
     """
-    return (isinstance(e, err.InterfaceError) and e.args[0] == "(0, '')") or\
-        (isinstance(e, err.OperationalError) and e.args[0] in operation_error_codes.values())
 
 
-class DataJointError(Exception):
+class QueryError(DataJointError):
     """
-    Base class for errors specific to DataJoint internal operation.
+    Errors arising from queries to the database
+    """
+
+
+# --- Third Level: QueryErrors ---
+class QuerySyntaxError(QueryError):
+    """
+    Errors arising from incorrect query syntax
+    """
+
+
+class AccessError(QueryError):
+    """
+    User access error: insufficient privileges.
+    """
+
+
+class UnknownAttributeError(DataJointError):
+    """
+    Error caused by referencing to a non-existing attributes
+    """
+
+
+class MissingTableError(DataJointError):
+    """
+    Query on a table that has not been declared
+    """
+
+
+class DuplicateError(QueryError):
+    """
+    An integrity error caused by a duplicate entry into a unique key
+    """
+
+
+class IntegrityError(QueryError):
+    """
+    An integrity error triggered by foreign key constraints
     """
-    pass
 
 
-class DuplicateError(DataJointError):
+class MissingAttributeError(QueryError):
     """
-    Error caused by a violation of a unique constraint when inserting data
+    An error arising when a required attribute value is not provided in INSERT
     """
-    pass
 
 
 class MissingExternalFile(DataJointError):

datajoint/fetch.py

@@ -121,7 +121,6 @@ def __call__(self, *attrs, offset=None, limit=None, order_by=None, format=None,
         :param download_path: for fetches that download data, e.g. attachments
         :return: the contents of the relation in the form of a structured numpy.array or a dict list
         """
-
         if order_by is not None:
             # if 'order_by' passed in a string, make into list
             if isinstance(order_by, str):

datajoint/jobs.py

@@ -2,7 +2,7 @@
 import os
 import platform
 from .table import Table
-from .errors import DuplicateError
+from .errors import DuplicateError, IntegrityError
 
 ERROR_MESSAGE_LENGTH = 2047
 TRUNCATION_APPENDIX = '...truncated'