Python操作MySQL数据库（非常详细）

MySQL 是目前最流行的开源关系数据库，大多应用于互联网行业。

例如，国内的百度、腾讯、淘宝、京东、网易、新浪等，国外的 Google、Facebook、Twitter、GitHub 等都在使用MySQL。社交、电商和游戏的核心存储往往也是 MySQL。

一、MySQL 简介

MySQL 具有体积小、速度快、开源等特点，许多中小型网站为了降低网站总体拥有成本而选择 MySQL 作为网站数据库。

MySQL 常用于如下场景：

1) Web 网站开发人员是 MySQL 最大的客户群，也是 MySQL 发展史上最为重要的支撑力量。

MySQL 之所以能成为 Web 网站开发人员最青睐的数据库管理系统，是因为 MySQL 数据库的安装配置非常简单，使用过程中的维护也不像很多大型商业数据库管理系统那么复杂，而且性能出色。

还有一个非常重要的原因就是 MySQL 是开源的，可以免费使用。

2) 嵌入式环境对软件系统最大的限制是硬件资源非常有限，在嵌入式环境下运行的软件系统，必须是轻量级低消耗的软件。

MySQL 在资源使用方面的伸缩性非常大，可以在资源非常充裕的环境下运行，也可以在资源非常少的环境下正常运行。对于嵌入式环境，MySQL 是一种非常合适的数据库系统，而且 MySQL 有专门针对嵌入式环境的版本。

二、MySQL 的使用

1. 常用数据类型

MySQL 支持的数据类型非常多，选择正确的数据类型对于获得高性能至关重要，下面介绍几种常用的数据类型。

1) 整数类型

整数类型有 TINYINT、SMALLINT、MEDIUMINT、INT 和 BIGINT 这几种，用于存储整数，每种类型的存储空间和值的范围不一样，我们需要根据实际情况选择合适的类型。

2) 实数类型

实数是带有小数部分的数字。FLOAT 和 DOUBLE 类型支持使用标准的浮点运算进行近似计算。DECIMAL 类型用于存储精确的小数。

在程序开发时，如果需要对小数进行精确计算，可以考虑使用 BIGINT 代替 DECIMAL，将需要存储的数值根据小数的位数乘相应的倍数即可。这样可以避免浮点存储计算不精确和 DECIMAL 精确计算代价高的问题。

3) 字符串类型

MySQL 支持多种字符串类型，如 VARCHAR、CHAR、BLOB、TEXT 等。

VARCHAR 和 CHAR 是两种最主要的字符串类型；BLOB 和 TEXT 是为存储大量数据而设计的字符串数据类型，分别采用二进制和字符方式存储。

VARCHAR用于存储可变长字符串，是最常见的字符串数据类型。相对于定长类型 CHAR，它更节省空间。

在存储数据时，VARCHAR 需要使用额外的字节记录字符串的长度：如果列的最大长度小于或等于 255 字节，则只使用1字节来记录，否则使用2字节。在 5.0 或者更高版本，MySQL 在存储和检索时会保留末位空格，并且长度按字符展示，如 varchar(20)，指的是 20 字符。

CHAR 是定长的，MySQL 会根据定义的字符串长度分配足够的空间。在存储 CHAR 值时，MySQL 会删除所有的末尾空格。CHAR 适合存储短字符串，或者接近同一个长度的值，如身份证号、手机号、电话等。

对于经常变更的数据，CHAR 比 VARCHAR 更好用，因为定长的 CHAR 不容易产生碎片。对于非常短的列，CHAR 比 VARCHAR 在存储空间上更有优势。例如，用 CHAR(1) 来存储只有 Y 和 N 的值，如果采用单字节字符集则只需要1字节，但是 VARCHAR(1) 却需要2字节，因为还有一个记录长度的额外字节。

4) 日期和时间类型

DATETIME 的时间范围是 1001 年到 9999 年，精度为秒。它使用 8 字节的存储空间，把日期和时间封装到格式为 YYYYMMDDHHMMSS 的整数中，与时区无关。

TIMESTAMP 只使用 4 字节的存储空间，因此它的范围比 DATETIME 小得多，只能表示 1970 年到 2038 年，并且 TIMESTAMP 会根据时区变化，具有自动更新能力。

当需要存储比秒更小粒度的日期和时间值时，虽然 MySQL 目前没有提供合适的数据类型，但是可以使用 BIGINT 存储微秒级别的时间戳，或者使用 DOUBLE 存储秒之后的小数部分。

2. 表结构的设计原则

在进行表结构设计时，需要牢记如下几个原则。

1) 更小的通常更好

在确保满足存储要求的前提下，尽量使用最小数据类型。因为它们占用更少的磁盘、内存和 CPU 缓存，并且处理时需要的 CPU 周期也更短，所以通常效率也更快。

2) 简单就好

简单数据类型的操作通常需要更短的 CPU 周期。例如，整形比字符操作代价更低；应该使用 MySQL 内置的类型（date、time和datetime）而不是字符串来存储日期和时间；应该使用整型存储 IP 地址。

3) 尽量避免使用NULL

NULL 是列的默认属性，但是通常情况下最好指定列为 NOT NULL。因为使用 NULL 的列在索引时比较复杂，当可使用 NULL 的列被索引时，每个索引记录需增加一个额外的字节。

3. 常用方法

在 Python 中，主要通过 PyMySQL 库进行 MySQL 数据库的操作。在使用 PyMySQL 库过程中，主要涉及连接对象和游标对象，大部分的操作都是基于这两个对象展开的。

我们先创建数据库连接对象（db），创建方法如下：

db = pymysql.connect(参数列表)

其中，参数列表中的参数如表1所示。

表1：pymysql.connect 的参数列表

参数	含义
host	主机地址（本地地址为 localhost）
port	端口号，默认 3306
user	用户名
password	密码
database	要操控的库
charset	编码方式，推荐使用 UTF-8

数据库连接对象（db）的常用方法如表2所示。

表2：数据库连接对象的常用方法

方法	描述
db.close()	关闭连接
db.commit()	提交数据库执行
db.rollback()	回滚到错误的语句执行之前的状态
cur = db.cursor()	返回游标对象，用于执行具体的 SQL 命令

游标对象（cur）的常用方法如表3所示。

表3：游标对象的常用方法

方法	描述
cur.execute(sql,[列表])	执行 SQL 命令，将查找结果存入游标对象 cur 中
cur.close()	关闭游标对象
cur.fetchone()	获取查询结果集的第一条数据
cur.fetchmany(n)	获取 n 条数据
cur.fetchall()	获取所有记录
cur.rowcount	返回查询结果记录数

4. 使用说明

本教程使用的 PyMySQL 1.0.2，PyMySQL 的使用分为如下几步。

1. 与数据库服务器建立连接：conn=pymysql.Connect(...)。
2. 获取游标对象（用于发送和接收数据）：cursor=conn.cursor()。
3. 使用游标执行 SQL 语句：cursor.excute(sql)。此时返回的是执行该语句后数据库表中受影响的数据条数。
4. 使用 fetch() 方法来获取执行的结果。
5. 关闭连接：先关闭游标，再关闭连接。

下面进行基本操作的示例演示。

1) 新建数据库：

import pymysql
　
con =pymysql.connect(host='localhost',user='root',passwd='root',port=3306,db='business')
cursor = con.cursor()
cursor.execute('create database if not exists business default charset utf8')
con.commit()

2) 新建表：

cursor.execute('use business')
cursor.execute('''CREATE table if not exists boss(id int auto_increment primary key,
           name varchar(20) not null,
           salary int not null)''')

3) 增加数据：

sql = """INSERT into boss(name,salary)
      values ('Jack',91), ('Harden',1300), ('Pony',200)"""
cursor.execute(sql)
con.commit()

4) 删除数据：

cursor.execute('delete from boss where salary < 100')
con.commit() # 一定要提交，提交了语句才生效

5) 修改数据：

cursor.execute("UPDATE boss set salary = 2000 where name = 'Pony'")
con.commit()

6) 关闭数据库：

con.close()

注入问题

用 excute() 方法执行 SQL 语句的时候，必须使用参数化的方式，内部执行参数化生成的 SQL 语句，对特殊字符加\转义，避免注入 SQL 语句漏洞生成。

连接池

面对大量的 Web 请求和插入、查询请求，MySQL 连接会不稳定，出现以下错误：

Lost connection to MySQL server during query ([Errno 104] Connection reset by peer)

解决方法是采用连接池方式，在程序创建连接的时候从连接池中获取一个连接，不需要重新初始化连接，可以提升获取连接的速度。

DBUtils 是一套 Python 数据库工具类库，可以对非线程安全的数据库接口进行线程安全包装，可用于各种多线程环境。

DBUtils 提供两种外部接口：

• PersistentDB 提供线程专用的数据库连接，并自动管理连接；
• PooledDB 提供线程间可共享的数据库连接，并自动管理连接。

三、封装示例

在实际数据库操作时，使用完毕后须要关闭游标（cursor）和连接（connection），为了简化代码，获取数据库连接采用 with open as 方式，执行完毕后自动关闭连接，无须主动关闭。

代码清单1中封装和使用 with 优化操作代码。采用 with 的方式来增加一个上下文管理器，并记录每次执行 SQL 预计的耗时。除了更好的采用 with 方式管理连接外，还简要封装了增删改查的操作。

代码清单1：mysqlUtil.py

# -*- coding: utf-8 -*-
# @Time : 2023/7/27 4:41 下午
# @Project : mysqlUtil
# @File : mysqlUtil.py
# @Version: Python3.9.8
　
import pymysql
from timeit import default_timer
　
host = 'localhost'
port = 3306
db = 'mysql_test'
user = 'mysql_test'
password = 'mysql_test'
　
# 用PyMySQL操作数据库
def get_connection():
    conn = pymysql.connect(host=host, port=port, db=db, user=user, password=password)
    return conn
　
# 使用with优化代码
class UsingMysql(object):
    def __init__(self, commit=True, log_time=True, log_label='总用时'):
        """
        :param commit: 是否在最后提交事务(设置为False的时候方便单元测试)
        :param log_time: 是否打印程序运行总时间
        :param log_label: 自定义log的文字
        """
        self._log_time = log_time
        self._commit = commit
        self._log_label = log_label
　
    def __enter__(self):
        # 如果需要记录时间
        if self._log_time is True:
            self._start = default_timer()
        # 在进入的时候自动获取连接和cursor
        conn = get_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        conn.autocommit = False
        self._conn = conn
        self._cursor = cursor
        return self
　
    def __exit__(self, *exc_info):
        # 提交事务
        if self._commit:
            self._conn.commit()
        # 在退出的时候自动关闭连接和cursor
        self._cursor.close()
        self._conn.close()
　
        if self._log_time is True:
            diff = default_timer() - self._start
            print('-- %s: %.6f 秒' % (self._log_label, diff))
　
    # 一系列封装的业务方法
    # 返回count
    def get_count(self, sql, params=None, count_key='count(id)'):
        self.cursor.execute(sql, params)
        data = self.cursor.fetchone()
        if not data:
            return 0
        return data[count_key]
　
    def fetch_one(self, sql, params=None):
        self.cursor.execute(sql, params)
        return self.cursor.fetchone()
　
    def fetch_all(self, sql, params=None):
        self.cursor.execute(sql, params)
        return self.cursor.fetchall()
　
    def fetch_by_pk(self, sql, pk):
        self.cursor.execute(sql, (pk,))
        return self.cursor.fetchall()
　
    def update_by_pk(self, sql, params=None):
        self.cursor.execute(sql, params)
　
    @property
    def cursor(self):
        return self._cursor
　
def check_it():
    """
    """
    with UsingMysql(log_time=True) as um:
        um.cursor.execute("select count(id) as total from Product")
        data = um.cursor.fetchone()
        print("-- 当前数量: %d " % data['total'])
　

if __name__ == '__main__':
    check_it()

运行上述代码，输出结果如下：
-- 当前数量: 0
-- 用时: 0.002345 秒

用这种方式改写代码之后，业务方法更精简，并且加入参数方便进行单元测试和监控代码的运行时间。