Python pymysql 模块，err() 实例源码

def make_connection(args):
    """
    Factory function to create and return a connection object.

    `args` is the parsed result from main arguments parser.
    """

    try:
        return pymysql.connect(
            host=args.host,
            port=args.port,
            user=args.user,
            password=args.password,
            db='igem',
            charset='utf8mb4',
            cursorclass=pymysql.cursors.DictCursor
        )
    except pymysql.err.DatabaseError as e:
        _display_error(e)

def init_db():
    connection = pymysql.connect(user='root')
    with connection.cursor() as c:
        try:
            c.execute('drop database hurricane')
        except pymysql.err.InternalError:
            # if database hurricane does not exist, just skip
            pass
        c.execute('create database hurricane')
    connection = pymysql.connect(user='root', db='hurricane')
    with connection.cursor() as c:
        c.execute('CREATE TABLE User (user_id INTEGER PRIMARY KEY AUTO_INCREMENT, email TEXT, name TEXT, password TEXT, bio TEXT)')
        connection.commit()
    connection.close()

def _close_conn(self):
        from pymysql.err import Error
        try:
            self.conn.close()
        except Error:
            pass

def setUpClass(cls):
        cls.setup_driver()

        # test connection
        try:
            cls.connect()
        except cls.driver.err.OperationalError:
            raise nose.SkipTest(
                "{0} - can't connect to MySQL server".format(cls))

def setUp(self):
        try:
            self.conn = self.connect()
        except self.driver.err.OperationalError:
            raise nose.SkipTest("Can't connect to MySQL server")

        self.pandasSQL = sql.SQLiteDatabase(self.conn, 'mysql')

        self._load_iris_data()
        self._load_test1_data()

def prepare_table(args):
    """
    To ensure the table structure was prepared.
    """

    with connection.cursor() as cursor:
        try:
            print('Checking if `parts_filtered` exists...')
            cursor.execute("DESCRIBE parts_filtered")
            structures = cursor.fetchall()
        except pymysql.err.ProgrammingError as e:
            existence = False
            print('`parts_filtered` does not exist')
        else:
            existence = True
            print('`parts_filtered` exists')

        if not existence or args.force_rebuild_table:
            print(
                'Rebuilding `parts_filtered` due to {}...'.format(
                    'non-existence'
                    if not existence
                    else '`force-rebuild-table` flag'
                )
            )

            print('Reinstalling stored procedure...')
            return_code = os.system(' '.join([
                'mysql',
                '-u{}'.format(args.user),
                '-p{}'.format(args.password) if args.password else '',
                '<',
                make_path('..', 'sql', 'igem', 'preprocess.sql')
            ]))
            if return_code != 0:
                print('Stored procedure installation failed with exit code {}'.format(return_code))
                sys.exit(return_code)

            print('Stored procedure installed')
            cursor.execute('CALL filter_parts')
        else:
            fields = {'ruler', 'ac'}
            wrong_types = [x['Field'] for x in structures if x['Field'] in fields and x['Type'] != 'longtext']

            if wrong_types:
                print("Fields %s have wrong types, dropping..." % ', '.join(wrong_types))
                for field in wrong_types:
                    cursor.execute('ALTER TABLE parts_filtered DROP COLUMN %s' % field)

            missing = fields - set(x['Field'] for x in structures)

            if missing:
                print('Fields %s missing, altering table...' % ', '.join(missing))
                for field in missing:
                    cursor.execute("ALTER TABLE parts_filtered ADD COLUMN %s longtext" % field)

    connection.commit()
    print('Table structure prepared')