Python 中连接 MySQL 服务器有两个库，分别为 PyMySQL 和 MySQLdb，
PyMySQL 是在 Python 3.x 版本中用于连接 MySQL 服务器的一个库，MySQLdb 只支持到 Python3.4，如果是 Python3.5+ 以上的版本，安装时会报缺少包。
两者使用方法一样，即使在 Python3.4 以下版本中安装 MySQLdb 也总是报缺少各种包，网上查询的解决方法也不可以，PyMySQL 既支持 Python 2 也支持 Python 3，安装简单方便，所以建议使用 PyMySQL
安装命令如下
pip install PyMySQL有些之前的项目中依然使用的是MySQLdb库，可以使用如下命令安装
pip install MySQL-Python --user --global-option=build_ext --global-option="-I/usr/local/opt/openssl/include" --global-option="-L/usr/local/opt/openssl/lib"笔者代码中使用了 logging 日志模块，没有安装的使用如下命令安装即可
pip install logging代码：
#!/usr/bin/python # -*- coding: UTF-8 -*- import logging import pymysql import pytz import time import traceback """ 日志配置 level=日志等级 datefmt=时间格式 format=日志格式 """ logging.basicConfig( level=logging.INFO, datefmt='%Y-%m-%d %H:%M:%S', format='%(asctime)s [%(levelname)s] %(message)s' ) class Constants(object): TIMEZONE = pytz.timezone('Asia/Shanghai') MYSQL_KEY_HOST = "host" MYSQL_KEY_USER = "user" MYSQL_KEY_PASSWORD = "password" MYSQL_KEY_PORT = "port" MYSQL_KEY_DATABASE = "database" class MySQLDBHelper(object): """ 初始化连接 """ def __init__(self, config): host = config[Constants.MYSQL_KEY_HOST] if host is None: raise ValueError("Parameter [host] is None.") port = config[Constants.MYSQL_KEY_PORT] if port is None: raise ValueError("Parameter [port] is None.") username = config[Constants.MYSQL_KEY_USER] if username is None: raise ValueError("Parameter [username] is None.") password = config[Constants.MYSQL_KEY_PASSWORD] if password is None: raise ValueError("Parameter [password] is None.") database = config[Constants.MYSQL_KEY_DATABASE] if database is None: raise ValueError("Parameter [database] is None.") logging.debug("[%s] 数据库初始化>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>开始" % database) start = time.time() self.host = host self.username = username self.password = password self.port = port self.database = database self.conn = None self.cursor = None try: self.conn = pymysql.connect(**config) self.cursor = self.conn.cursor() end = time.time() logging.debug("[%s] 数据库初始化>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>结束" % database) logging.info("[%s] 数据库初始化成功。耗时：%d ms。" % (database, (end - start))) except Exception as e: logging.error("数据库初始化失败!") logging.error(traceback.format_exc()) """ 释放资源 """ def close(self): try: if not self.conn: self.conn.close() except Exception as e: logging.error("数据库连接释放异常!") logging.error(traceback.format_exc()) """ 创建表 """ def create_table(self, sql): try: logging.info("[%s] SQL >>> [%s]" % (self.database, sql)) self.cursor.execute(sql) self.conn.commit() except Exception as e: logging.error(traceback.format_exc()) """ 执行查询 SQL 语句 :param str sql: 查询sql :param bool single: 是否查询单个结果集，默认False :return: 返回查询数据 :rtype: list tuple [(e1, e2, ...), (e1, e2, ...), ...] """ def execute_query(self, sql, single=False): try: logging.info("[%s] SQL >>> [%s]" % (self.database, sql)) self.cursor.execute(sql) if single: result_tuple = self.cursor.fetchone() else: result_tuple = self.cursor.fetchall() return result_tuple except Exception as e: logging.error(traceback.format_exc()) """ 执行 SQL 语句 :param str sql: 插入、删除、修改 :return: 返回受影响行数 :rtype: int """ def execute_commit(self, sql): try: logging.info("[%s] SQL >>> [%s]" % (self.database, sql)) affect_rows = self.cursor.execute(sql) self.conn.commit() return affect_rows except Exception as e: self.conn.rollback() logging.error(traceback.format_exc()) """ 批量插入 :param str sql: 插入 :return: 返回受影响行数 :rtype: int """ def insert_many(self, sql, values): try: affect_rows = self.cursor.executemany(sql, values) self.conn.commit() return affect_rows except Exception as e: self.conn.rollback() logging.error("[%s] SQL >>> [%s] param >>> %s" % (self.database, sql, values)) logging.error(traceback.format_exc()) if __name__ == '__main__': # 初始化连接 config = { Constants.MYSQL_KEY_HOST: '127.0.0.1', Constants.MYSQL_KEY_PORT: 3306, Constants.MYSQL_KEY_USER: 'root', Constants.MYSQL_KEY_PASSWORD: '123456', Constants.MYSQL_KEY_DATABASE: 'test_db' } db = MySQLDBHelper(config=config) # 创建表 create_table_sql = """ CREATE TABLE IF NOT EXISTS `students` ( `id` int(11) NOT NULL AUTO_INCREMENT, `name` varchar(50) NOT NULL, `age` int(11) NOT NULL, `gender` int(11) NOT NULL, `address` varchar(100) NOT NULL, `email` varchar(50) NOT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8""" db.create_table(create_table_sql) # 插入数据 name = "张三" age = 18 gender = 1 address = "北京朝阳" email = "1@qq.com" insert_sql = """insert into students(`name`, `age`, `gender`, `address`, `email`) values ('%s', %d, %d, '%s', '%s')""" % (name, age, gender, address, email) db.execute_commit(insert_sql) # 批量插入数据 insert_many_sql = """insert into students(`name`, `age`, `gender`, `address`, `email`) values (%s, %s, %s, %s, %s)""" values = [ ("李丽", 20, 0, "北京海淀", "2@qq.com"), ("李四", 21, 1, "天津南开", "3@qq.com"), ("孙俪", 19, 0, "北京东城", "4@qq.com"), ("赵钱", 22, 1, "辽宁大连", "5@qq.com"), ("王五", 20, 1, "河北石家庄", "6@qq.com") ] db.insert_many(insert_many_sql, values) # 更新数据 name = "张三" age = 22 address = "河北保定" update_sql = """update students set `age` = %d, `address` = '%s' where `name` = '%s'""" % (age, address, name) db.execute_commit(update_sql) # 删除数据 name = "张三" delete_sql = """delete from students where `name` = '%s'""" % name db.execute_commit(delete_sql) # 释放连接 db.close()运行结果：
2019-10-31 19:20:39 [INFO] [test_db] 数据库初始化成功。耗时：0 ms。 2019-10-31 19:20:39 [INFO] [test_db] SQL >>> [ CREATE TABLE IF NOT EXISTS `students` ( `id` int(11) NOT NULL AUTO_INCREMENT, `name` varchar(50) NOT NULL, `age` int(11) NOT NULL, `gender` int(11) NOT NULL, `address` varchar(100) NOT NULL, `email` varchar(50) NOT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8] 2019-10-31 19:20:39 [INFO] [test_db] SQL >>> [insert into students(`name`, `age`, `gender`, `address`, `email`) values ('张三', 18, 1, '北京朝阳', '1@qq.com')] 2019-10-31 19:20:39 [INFO] [test_db] SQL >>> [update students set `age` = 22, `address` = '河北保定' where `name` = '张三'] 2019-10-31 19:20:39 [INFO] [test_db] SQL >>> [delete from students where `name` = '张三']