MySQL for Python

--- a/test_MySQLdb_dbapi20.py
+++ b/MySQLdb/exceptions.py
@@ -1,205 +1,117 @@
-#!/usr/bin/env python
-import dbapi20
-import unittest
-import MySQLdb
+"""
+MySQLdb.exceptions
+==================
 
-class test_MySQLdb(dbapi20.DatabaseAPI20Test):
-    driver = MySQLdb
-    connect_args = ()
-    connect_kw_args = dict(db='test',
-                           read_default_file='~/.my.cnf',
-                           charset='utf8',
-                           sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL")
+These classes are dictated by the DB API v2.0:
 
-    def test_setoutputsize(self): pass
-    def test_setoutputsize_basic(self): pass
-    def test_nextset(self): pass
+    http://www.python.org/topics/database/DatabaseAPI-2.0.html
+"""
 
-    """The tests on fetchone and fetchall and rowcount bogusly
-    test for an exception if the statement cannot return a
-    result set. MySQL always returns a result set; it's just that
-    some things return empty result sets."""
+# from __future__ import absolute_import
+# Unfortunately, you cannot put the above in a conditional statement.
+# It would make things much cleaner for Python-2.5, but breaks older.
+
+try:
+    from exceptions import Exception, StandardError, Warning
+except ImportError:
+    import sys
+    e = sys.modules['exceptions']
+    StandardError = e.StandardError
+    Warning = e.Warning
     
-    def test_fetchall(self):
-        con = self._connect()
-        try:
-            cur = con.cursor()
-            # cursor.fetchall should raise an Error if called
-            # without executing a query that may return rows (such
-            # as a select)
-            self.assertRaises(self.driver.Error, cur.fetchall)
+from MySQLdb.constants import ER
 
-            self.executeDDL1(cur)
-            for sql in self._populate():
-                cur.execute(sql)
+class MySQLError(StandardError):
+    
+    """Exception related to operation with MySQL."""
 
-            # cursor.fetchall should raise an Error if called
-            # after executing a a statement that cannot return rows
-##             self.assertRaises(self.driver.Error,cur.fetchall)
 
-            cur.execute('select name from %sbooze' % self.table_prefix)
-            rows = cur.fetchall()
-            self.failUnless(cur.rowcount in (-1,len(self.samples)))
-            self.assertEqual(len(rows),len(self.samples),
-                'cursor.fetchall did not retrieve all rows'
-                )
-            rows = [r[0] for r in rows]
-            rows.sort()
-            for i in range(0,len(self.samples)):
-                self.assertEqual(rows[i],self.samples[i],
-                'cursor.fetchall retrieved incorrect rows'
-                )
-            rows = cur.fetchall()
-            self.assertEqual(
-                len(rows),0,
-                'cursor.fetchall should return an empty list if called '
-                'after the whole result set has been fetched'
-                )
-            self.failUnless(cur.rowcount in (-1,len(self.samples)))
+class Warning(Warning, MySQLError):
 
-            self.executeDDL2(cur)
-            cur.execute('select name from %sbarflys' % self.table_prefix)
-            rows = cur.fetchall()
-            self.failUnless(cur.rowcount in (-1,0))
-            self.assertEqual(len(rows),0,
-                'cursor.fetchall should return an empty list if '
-                'a select query returns no rows'
-                )
-            
-        finally:
-            con.close()
-                
-    def test_fetchone(self):
-        con = self._connect()
-        try:
-            cur = con.cursor()
+    """Exception raised for important warnings like data truncations
+    while inserting, etc."""
 
-            # cursor.fetchone should raise an Error if called before
-            # executing a select-type query
-            self.assertRaises(self.driver.Error,cur.fetchone)
+class Error(MySQLError):
 
-            # cursor.fetchone should raise an Error if called after
-            # executing a query that cannnot return rows
-            self.executeDDL1(cur)
-##             self.assertRaises(self.driver.Error,cur.fetchone)
+    """Exception that is the base class of all other error exceptions
+    (not Warning)."""
 
-            cur.execute('select name from %sbooze' % self.table_prefix)
-            self.assertEqual(cur.fetchone(),None,
-                'cursor.fetchone should return None if a query retrieves '
-                'no rows'
-                )
-            self.failUnless(cur.rowcount in (-1,0))
 
-            # cursor.fetchone should raise an Error if called after
-            # executing a query that cannnot return rows
-            cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
-                self.table_prefix
-                ))
-##             self.assertRaises(self.driver.Error,cur.fetchone)
+class InterfaceError(Error):
 
-            cur.execute('select name from %sbooze' % self.table_prefix)
-            r = cur.fetchone()
-            self.assertEqual(len(r),1,
-                'cursor.fetchone should have retrieved a single row'
-                )
-            self.assertEqual(r[0],'Victoria Bitter',
-                'cursor.fetchone retrieved incorrect data'
-                )
-##             self.assertEqual(cur.fetchone(),None,
-##                 'cursor.fetchone should return None if no more rows available'
-##                 )
-            self.failUnless(cur.rowcount in (-1,1))
-        finally:
-            con.close()
+    """Exception raised for errors that are related to the database
+    interface rather than the database itself."""
 
-    # Same complaint as for fetchall and fetchone
-    def test_rowcount(self):
-        con = self._connect()
-        try:
-            cur = con.cursor()
-            self.executeDDL1(cur)
-##             self.assertEqual(cur.rowcount,-1,
-##                 'cursor.rowcount should be -1 after executing no-result '
-##                 'statements'
-##                 )
-            cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
-                self.table_prefix
-                ))
-##             self.failUnless(cur.rowcount in (-1,1),
-##                 'cursor.rowcount should == number or rows inserted, or '
-##                 'set to -1 after executing an insert statement'
-##                 )
-            cur.execute("select name from %sbooze" % self.table_prefix)
-            self.failUnless(cur.rowcount in (-1,1),
-                'cursor.rowcount should == number of rows returned, or '
-                'set to -1 after executing a select statement'
-                )
-            self.executeDDL2(cur)
-##             self.assertEqual(cur.rowcount,-1,
-##                 'cursor.rowcount not being reset to -1 after executing '
-##                 'no-result statements'
-##                 )
-        finally:
-            con.close()
 
-    def test_callproc(self):
-        pass # performed in test_MySQL_capabilities
+class DatabaseError(Error):
 
-    def help_nextset_setUp(self,cur):
-        ''' Should create a procedure called deleteme
-            that returns two result sets, first the 
-	    number of rows in booze then "name from booze"
-        '''
-        sql="""
-           create procedure deleteme()
-           begin
-               select count(*) from %(tp)sbooze;
-               select name from %(tp)sbooze;
-           end
-        """ % dict(tp=self.table_prefix)
-        cur.execute(sql)
+    """Exception raised for errors that are related to the
+    database."""
 
-    def help_nextset_tearDown(self,cur):
-        'If cleaning up is needed after nextSetTest'
-        cur.execute("drop procedure deleteme")
 
-    def test_nextset(self):
-        from warnings import warn
-        con = self._connect()
-        try:
-            cur = con.cursor()
-            if not hasattr(cur,'nextset'):
-                return
+class DataError(DatabaseError):
 
-            try:
-                self.executeDDL1(cur)
-                sql=self._populate()
-                for sql in self._populate():
-                    cur.execute(sql)
+    """Exception raised for errors that are due to problems with the
+    processed data like division by zero, numeric value out of range,
+    etc."""
 
-                self.help_nextset_setUp(cur)
 
-                cur.callproc('deleteme')
-                numberofrows=cur.fetchone()
-                assert numberofrows[0]== len(self.samples)
-                assert cur.nextset()
-                names=cur.fetchall()
-                assert len(names) == len(self.samples)
-                s=cur.nextset()
-                if s:
-                    empty = cur.fetchall()
-                    self.assertEquals(len(empty), 0,
-                                      "non-empty result set after other result sets")
-                    #warn("Incompatibility: MySQL returns an empty result set for the CALL itself",
-                    #     Warning)
-                #assert s == None,'No more return sets, should return None'
-            finally:
-                self.help_nextset_tearDown(cur)
+class OperationalError(DatabaseError):
 
-        finally:
-            con.close()
+    """Exception raised for errors that are related to the database's
+    operation and not necessarily under the control of the programmer,
+    e.g. an unexpected disconnect occurs, the data source name is not
+    found, a transaction could not be processed, a memory allocation
+    error occurred during processing, etc."""
 
-    
-if __name__ == '__main__':
-    unittest.main()
-    print '''"Huh-huh, he said 'unit'." -- Butthead'''
+
+class IntegrityError(DatabaseError):
+
+    """Exception raised when the relational integrity of the database
+    is affected, e.g. a foreign key check fails, duplicate key,
+    etc."""
+
+
+class InternalError(DatabaseError):
+
+    """Exception raised when the database encounters an internal
+    error, e.g. the cursor is not valid anymore, the transaction is
+    out of sync, etc."""
+
+
+class ProgrammingError(DatabaseError):
+
+    """Exception raised for programming errors, e.g. table not found
+    or already exists, syntax error in the SQL statement, wrong number
+    of parameters specified, etc."""
+
+
+class NotSupportedError(DatabaseError):
+
+    """Exception raised in case a method or database API was used
+    which is not supported by the database, e.g. requesting a
+    .rollback() on a connection that does not support transaction or
+    has transactions turned off."""
+
+
+error_map = {}
+
+def _map_error(exc, *errors):
+    for error in errors:
+        error_map[error] = exc
+
+_map_error(ProgrammingError, ER.DB_CREATE_EXISTS, ER.SYNTAX_ERROR,
+           ER.PARSE_ERROR, ER.NO_SUCH_TABLE, ER.WRONG_DB_NAME,
+           ER.WRONG_TABLE_NAME, ER.FIELD_SPECIFIED_TWICE,
+           ER.INVALID_GROUP_FUNC_USE, ER.UNSUPPORTED_EXTENSION,
+           ER.TABLE_MUST_HAVE_COLUMNS, ER.CANT_DO_THIS_DURING_AN_TRANSACTION)
+_map_error(DataError, ER.WARN_DATA_TRUNCATED, ER.WARN_NULL_TO_NOTNULL,
+           ER.WARN_DATA_OUT_OF_RANGE, ER.NO_DEFAULT, ER.PRIMARY_CANT_HAVE_NULL,
+           ER.DATA_TOO_LONG, ER.DATETIME_FUNCTION_OVERFLOW)
+_map_error(IntegrityError, ER.DUP_ENTRY, ER.NO_REFERENCED_ROW,
+           ER.NO_REFERENCED_ROW_2, ER.ROW_IS_REFERENCED, ER.ROW_IS_REFERENCED_2,
+           ER.CANNOT_ADD_FOREIGN)
+_map_error(NotSupportedError, ER.WARNING_NOT_COMPLETE_ROLLBACK,
+           ER.NOT_SUPPORTED_YET, ER.FEATURE_DISABLED, ER.UNKNOWN_STORAGE_ENGINE)
+
+del StandardError, _map_error, ER