网站首页 > 技术文章正文

技术干货|Python MySQL客户端库PyMySQL常见用法

hfteth 2025-06-24 13:43:10 技术文章 3 ℃

PyMySQL是一个用于Python的纯Python MySQL客户端库，它实现了PEP 249 Python数据库API规范，并提供了对MySQL数据库的连接和操作功能。

代码示例：

连接到MySQL数据库：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 关闭连接
conn.close()

执行SQL查询语句：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行带参数的SQL查询语句：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行带参数的SQL查询语句
sql = "SELECT * FROM users WHERE age > %s"
cursor.execute(sql, (18,))

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL插入语句：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL插入语句
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
cursor.execute(sql, ('John', 25))

# 提交事务
conn.commit()

# 关闭连接
conn.close()

执行SQL更新语句：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL更新语句
sql = "UPDATE users SET age = %s WHERE id = %s"
cursor.execute(sql, (30, 1))

# 提交事务
conn.commit()

# 关闭连接
conn.close()

执行SQL删除语句：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL删除语句
sql = "DELETE FROM users WHERE id = %s"
cursor.execute(sql, (1,))

# 提交事务
conn.commit()

# 关闭连接
conn.close()

获取最后插入的自增ID：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL插入语句
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
cursor.execute(sql, ('John', 25))

# 获取最后插入的自增ID
last_insert_id = cursor.lastrowid

# 提交事务
conn.commit()

# 关闭连接
conn.close()

# 打印最后插入的自增ID
print(last_insert_id)

使用事务处理：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 开始事务
conn.begin()

try:
    # 创建游标对象
    cursor = conn.cursor()

    # 执行SQL语句
    sql1 = "INSERT INTO users (name, age) VALUES (%s, %s)"
    cursor.execute(sql1, ('John', 25))

    sql2 = "UPDATE users SET age = %s WHERE id = %s"
    cursor.execute(sql2, (30, 1))

    # 提交事务
    conn.commit()
except:
    # 回滚事务
    conn.rollback()

# 关闭连接
conn.close()

执行多条SQL语句：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行多条SQL语句
sql = """
    INSERT INTO users (name, age) VALUES ('John', 25);
    UPDATE users SET age = 30 WHERE id = 1;
"""
cursor.execute(sql)

# 提交事务
conn.commit()

# 关闭连接
conn.close()

执行SQL查询并分页：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句并分页
sql = "SELECT * FROM users LIMIT %s, %s"
cursor.execute(sql, (0, 10))

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

使用字典类型获取查询结果：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb', cursorclass=pymysql.cursors.DictCursor)

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行存储过程：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行存储过程
cursor.callproc('get_users')

# 获取输出参数值
output_param = cursor.fetchone()
print(output_param)

# 关闭连接
conn.close()

执行事务中的多个SQL语句：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

try:
    # 开始事务
    conn.begin()

    # 创建游标对象
    cursor = conn.cursor()

    # 执行SQL语句
    sql1 = "INSERT INTO users (name, age) VALUES ('John', 25)"
    cursor.execute(sql1)

    sql2 = "UPDATE users SET age = 30 WHERE id = 1"
    cursor.execute(sql2)

    # 提交事务
    conn.commit()
except:
    # 回滚事务
    conn.rollback()

# 关闭连接
conn.close()

执行SQL查询并获取字段名：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取字段名
column_names = [desc[0] for desc in cursor.description]
print(column_names)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的行数：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的行数
num_rows = cursor.rowcount
print(num_rows)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的列数：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的列数
num_columns = len(cursor.description)
print(num_columns)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

使用连接池进行数据库连接：

import pymysql
from DBUtils.PooledDB import PooledDB

# 创建连接池
pool = PooledDB(pymysql, host='localhost', user='root', password='password', database='mydb')

# 从连接池获取连接
conn = pool.connection()

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

使用事务管理器进行数据库连接：

import pymysql
from pymysql import err

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 开始事务
conn.begin()

# 创建游标对象
cursor = conn.cursor()

try:
    # 执行SQL语句
    sql1 = "INSERT INTO users (name, age) VALUES ('John', 25)"
    cursor.execute(sql1)

    sql2 = "UPDATE users SET age = 30 WHERE id = 1"
    cursor.execute(sql2)

    # 提交事务
    conn.commit()
except err as e:
    # 回滚事务
    conn.rollback()

# 关闭连接
conn.close()

使用with语句进行数据库连接：

import pymysql

# 创建数据库连接
with pymysql.connect(host='localhost', user='root', password='password', database='mydb') as conn:
    # 创建游标对象
    with conn.cursor() as cursor:
        # 执行SQL查询语句
        sql = "SELECT * FROM users"
        cursor.execute(sql)

        # 获取查询结果
        results = cursor.fetchall()

        # 打印结果
        for row in results:
            print(row)

使用连接池进行数据库连接并设置连接池大小：

import pymysql
from DBUtils.PooledDB import PooledDB

# 创建连接池
pool = PooledDB(pymysql, host='localhost', user='root', password='password', database='mydb', maxconnections=10)

# 从连接池获取连接
conn = pool.connection()

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

设置字符集：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb', charset='utf8mb4')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

设置自动提交模式：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb', autocommit=True)

# 创建游标对象
cursor = conn.cursor()

# 执行SQL插入语句
sql = "INSERT INTO users (name, age) VALUES ('John', 25)"
cursor.execute(sql)

# 关闭连接
conn.close()

设置超时时间：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb', connect_timeout=10)

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

使用fetchone()方法获取查询结果的一行数据：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的一行数据
row = cursor.fetchone()

# 打印结果
print(row)

# 关闭连接
conn.close()

使用fetchmany()方法获取查询结果的多行数据：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的多行数据
rows = cursor.fetchmany(5)

# 打印结果
for row in rows:
    print(row)

# 关闭连接
conn.close()

使用scroll()方法滚动浏览查询结果：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 滚动浏览查询结果
cursor.scroll(2, mode='absolute')  # 绝对位置
row = cursor.fetchone()
print(row)

cursor.scroll(-1, mode='relative')  # 相对位置
row = cursor.fetchone()
print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的列名和行数据：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取字段名
column_names = [desc[0] for desc in cursor.description]
print(column_names)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的字典类型数据：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb', cursorclass=pymysql.cursors.DictCursor)

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的字典类型数据
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的生成器：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的生成器
results = cursor.fetchall()

# 生成器循环打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的元组类型数据：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的元组类型数据
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的列表类型数据：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 获取查询结果的列表类型数据
results = list(cursor.fetchall())

# 打印结果
for row in results:
    print(row)

# 关闭连接
conn.close()

执行SQL查询并获取查询结果的集合类型数据：

import pymysql

# 创建数据库连接
conn = pymysql.connect(host='localhost', user='root', password='password', database='mydb')

# 创建游标对象
cursor = conn.cursor(pymysql.cursors.DictCursor)

# 执行SQL查询语句
sql = "SELECT * FROM users"
cursor.execute(sql)

# 关闭连接
conn.close()

mysql其他操作可参考：

mysql8安全基线及安全优化

mysql8.0安装、部署及优化

Shell脚本来实现MySQL服务的定时健康性检查

shell脚本实现mysql数据库文件和日志备份

技术干货|MySQL 8索引优化、查询优化、表设计优化和配置优化

Ansible实现生产环境一键式Linux+Nginx+MySQL+Tomcat企业级架构

Python MySQL驱动模块mysql-connector-python常见用法