pyTestMysqlConnection.py 文件源码

def mysql_query(host, user, password, port, charset, database, sql):
    try:
        connection = pymysql.connect(host=host, user=user, password=password, database=database, port=port,
                                     charset=charset,
                                     cursorclass=pymysql.cursors.DictCursor, connect_timeout=5)
    except pymysql.Error as e:
        # raise RuntimeError("Can't connect to MySQL server")
        print e.message or e.args
        sys.exit(1)

    try:
        with connection.cursor() as cursor:
            cursor.execute(sql)
        connection.commit()
    finally:
        connection.close()
    if cursor is not None:
        return cursor
    else:
        sys.exit(1)