refactor error handling

Author: dimitri-yatsenko

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,49 @@
 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, args):
+    """
+    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
+    :param args: values for query 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(*args[1:], query)
+    if isinstance(client_error, client.err.InternalError) and client_error.args[0] == 1364:
+        return errors.MissingAttributeValueError(*args[1:])
+    raise client_error
+
 
 logger = logging.getLogger(__name__)
 
@@ -60,6 +97,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 +117,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 +141,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 +179,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, args)
+
+    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 +238,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")

datajoint/declare.py

@@ -210,8 +210,8 @@ def compile_foreign_key(line, context, attributes, primary_key, attr_sql, foreig
     # declare the foreign key
     foreign_key_sql.append(
         'FOREIGN KEY (`{fk}`) REFERENCES {ref} (`{pk}`) ON UPDATE CASCADE ON DELETE RESTRICT'.format(
-            fk='`,`'.join(ref.primary_key),
-            pk='`,`'.join(base.primary_key),
+            fk='`,`'.join(ref.primary_key) or '_',  # dimensionless tables use _ as primary key
+            pk='`,`'.join(base.primary_key) or '_',
             ref=base.full_table_name))
 
     # declare unique index
@@ -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
@@ -277,7 +277,9 @@ def declare(full_table_name, definition, context):
         definition, context)
 
     if not primary_key:
-        raise DataJointError('Table must have a primary key')
+        # singular (dimensionless) table -- can contain only one element
+        attribute_sql.insert(0, '`_` char(1) not null default "" COMMENT "dimensionless primary key"')
+        primary_key = ['_']
 
     return (
         'CREATE TABLE IF NOT EXISTS %s (\n' % full_table_name +

datajoint/errors.py

@@ -1,37 +1,72 @@
-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
     """
-    pass
 
 
-class DuplicateError(DataJointError):
+# --- 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
+    """
+
+class MissingAttributeValueError(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

datajoint/heading.py

@@ -182,6 +182,9 @@ def init_from_database(self, conn, database, table_name):
                        for k, v in x.items() if k not in fields_to_drop}
                       for x in attributes]
 
+        # exclude attributes that begin with an underscore: these are reserved for DataJoint internals
+        attributes = [a for a in attributes if not a['name'].startswith('_')]
+
         numeric_types = {
             ('float', False): np.float64,
             ('float', True): np.float64,

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'