PyMySQL数据库连接与优化方式_Mysql

本文将介绍如何使用 pymysql 连接和操作 mysql 数据库，包括基本连接、crud 操作、事务处理以及如何在高并发环境下使用连接池优化性能。

通过合理的连接池配置和错误处理机制，可以构建出稳定高效的数据库应用。

一、pymysql 简介

pymysql 是一个纯 python 实现的 mysql 客户端库，用于连接和操作 mysql 数据库。它完全兼容 python db api 2.0 规范，提供了简单易用的接口来执行 sql 查询和操作。

核心优势

纯 python 实现：无需外部依赖，跨平台兼容性好
python 3 全面支持：兼容最新 python 特性和语法
线程安全：支持多线程并发操作
完整功能支持：事务、存储过程、预处理语句等
广泛兼容：支持 mysql 5.5+ 和 mariadb

安装方法

pip install pymysql

二、数据库连接配置

基础连接方式

import pymysql
from pymysql.cursors import dictcursor

# 推荐配置方式
def create_connection():
    return pymysql.connect(
        host='localhost',      # 数据库地址
        user='username',       # 用户名
        password='password',   # 密码
        database='test_db',    # 数据库名
        port=3306,            # 端口，默认3306
        charset='utf8mb4',     # 字符集，推荐utf8mb4
        autocommit=false,     # 是否自动提交
        cursorclass=dictcursor # 返回字典格式结果
    )

连接参数说明

参数	说明	值
host	数据库服务器地址	‘localhost’
user	用户名	根据实际配置
password	密码	根据实际配置
database	数据库名称	项目数据库名
charset	字符编码	‘utf8mb4’（支持表情符号）
autocommit	自动提交事务	false（建议手动控制）
cursorclass	游标类型	dictcursor（结果以字典返回）

cursorclass参数说明

cursorclass	说明	返回结果格式	适用场景
cursor (默认)	普通游标	元组格式 (value1, value2, …)	基础查询，需要最高性能时
dictcursor	字典游标	字典格式 {‘column’: value}	需要按列名访问数据时
sscursor	无缓冲游标	元组格式，流式读取	处理大量数据，内存有限时
ssdictcursor	无缓冲字典游标	字典格式，流式读取	大量数据且需要按列名访问
cursor 子类	自定义游标	自定义格式	特殊数据处理需求

完整连接示例

import pymysql
from pymysql.cursors import dictcursor

def get_db_connection():
    """获取数据库连接"""
    return pymysql.connect(
        host='localhost',
        user='myuser',
        password='mypassword',
        database='mydatabase',
        charset='utf8mb4',
        autocommit=false,
        cursorclass=dictcursor,
        connect_timeout=10  # 连接超时10秒
    )

# 使用示例
def test_connection():
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("select 1 as test")
            result = cursor.fetchone()
            print("连接测试成功:", result)
    finally:
        conn.close()

test_connection()

输出：

连接测试成功: {'test': 1}

三、数据库基础操作

创建示例数据表

create table mydb.users (
    id int auto_increment primary key,
    name varchar(100) not null,
    email varchar(100) unique not null,
    age int,
    created_at timestamp default current_timestamp
);

相关说明

关键字	类型	说明
int	数据类型	整数类型，用于存储整数值
auto_increment	约束/属性	自动递增，每次插入新记录时自动生成唯一id
primary key	约束	主键，唯一标识每条记录
varchar(100)	数据类型	可变长度字符串，最大100字符
not null	约束	该字段不能为空，必须包含值
unique	约束	确保每个值唯一，不允许重复
timestamp	数据类型	时间戳类型，用于存储日期和时间
default current_timestamp	默认值	默认值为当前系统时间

数据库操作封装类

import pymysql
from pymysql.cursors import dictcursor
from typing import list, dict, any, optional, tuple

class mysqlmanager:
    """mysql 数据库管理类"""

    def __init__(self, config: dict[str, any]):
        self.config = config

    def execute_query(self, sql: str, params: tuple = none) -> list[dict]:
        """执行查询语句（select）"""
        conn = pymysql.connect(**self.config)
        try:
            with conn.cursor(dictcursor) as cursor:
                cursor.execute(sql, params or ())
                return cursor.fetchall()
        finally:
            conn.close()

    def execute_update(self, sql: str, params: tuple = none) -> int:
        """执行更新语句（insert/update/delete）"""
        conn = pymysql.connect(**self.config)
        try:
            with conn.cursor() as cursor:
                affected_rows = cursor.execute(sql, params or ())
                conn.commit()
                return affected_rows
        except exception as e:
            conn.rollback()
            raise e
        finally:
            conn.close()

crud 操作示例

操作	英文	中文	对应 sql	描述
c	create	创建	insert	创建新记录
r	read	读取	select	查询/读取数据
u	update	更新	update	修改现有记录
d	delete	删除	delete	删除记录

# 数据库配置
db_config = {
    'host': 'localhost',
    'user': 'root',
    'password': 'password',
    'database': 'test_db',
    'charset': 'utf8mb4',
    'cursorclass': dictcursor 
}

db = mysqlmanager(db_config)

# 1. 插入数据
def add_user(name: str, email: str, age: int) -> int:
    sql = "insert into users (name, email, age) values (%s, %s, %s)"
    return db.execute_update(sql, (name, email, age))

# 2. 查询数据
def get_all_users() -> list[dict]:
    return db.execute_query("select * from users")

# 3. 更新数据
def update_user_email(user_id: int, new_email: str) -> int:
    sql = "update users set email = %s where id = %s"
    return db.execute_update(sql, (new_email, user_id))

# 4. 删除数据
def delete_user(user_id: int) -> int:
    return db.execute_update("delete from users where id = %s", (user_id,))
	
if __name__ == '__main__':
    users = get_all_users()
    print(f"查询所有用户: {users}")
	
    user_id = add_user("张三", "zhangsan@example.com", 25)
    print(f"执行：插入新用户")
	
    users = get_all_users()
    print(f"查询所有用户: {users}")
	
    user_id = users[0]['id']
    affected_rows = update_user_email(user_id, "zhangsan2@example.com")
    print(f"执行：更新邮箱，影响行数: {affected_rows}")
	
    users = get_all_users()
    print(f"查询所有用户: {users}")
	
    affected_rows = delete_user(user_id)
    print(f"执行：删除用户，影响行数: {affected_rows}")
	
    users = get_all_users()
    print(f"查询所有用户: {users}")

输出：

查询所有用户: ()
执行：插入新用户
查询所有用户: [{'id': 3, 'name': '张三', 'email': 'zhangsan@example.com', 'age': 25, 'created_at': datetime.datetime(2025, 9, 23, 19, 25, 11)}]
执行：更新邮箱，影响行数: 1
查询所有用户: [{'id': 3, 'name': '张三', 'email': 'zhangsan2@example.com', 'age': 25, 'created_at': datetime.datetime(2025, 9, 23, 19, 25, 11)}]
执行：删除用户，影响行数: 1
查询所有用户: ()

事务处理示例

模拟简单的转账操作，从一个用户账户转移到另一个用户账户。

def transfer_points(sender_id: int, receiver_id: int, points: int) -> bool:
    """转账操作（事务示例）"""
    conn = pymysql.connect(**db_config)
    try:
        with conn.cursor(dictcursor) as cursor:
            # 检查发送者余额
            cursor.execute("select points from accounts where user_id = %s", (sender_id,))
            sender = cursor.fetchone()
            
            if not sender or sender['points'] < points:
                raise valueerror("余额不足")
            
            # 执行转账
            cursor.execute("update accounts set points = points - %s where user_id = %s", 
                         (points, sender_id))
            cursor.execute("update accounts set points = points + %s where user_id = %s", 
                         (points, receiver_id))
            
            conn.commit()
            return true
    except exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

批量操作

def batch_insert_users(users: list[tuple]) -> int:
    """批量插入用户数据"""
    sql = "insert into users (name, email, age) values (%s, %s, %s)"
    conn = pymysql.connect(**db_config)
    try:
        with conn.cursor() as cursor:
            affected_rows = cursor.executemany(sql, users)
            conn.commit()
            return affected_rows
    except exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

# 使用示例
users_data = [
    ('张三', 'zhangsan@example.com', 25),
    ('李四', 'lisi@example.com', 30)
]
batch_insert_users(users_data)

四、连接池优化

为什么需要连接池

频繁创建和关闭数据库连接会导致：

资源浪费（tcp 连接建立开销）
性能下降（连接初始化时间）
连接数耗尽（超过数据库最大连接数）
连接池通过复用连接解决这些问题。

使用 dbutils 实现连接池

安装方法

pip install dbutils

实现示例

from dbutils.pooled_db import pooleddb
import pymysql
import threading
from typing import list, dict, any, tuple
from pymysql.cursors import dictcursor

class connectionpool:
    """数据库连接池"""
    
    _instance = none
    _lock = threading.lock()
    
    def __new__(cls, config: dict[str, any]):
        with cls._lock:
            if cls._instance is none:
                cls._instance = super().__new__(cls)
                cls._instance.pool_config = config.copy()
                cls._instance._pool = pooleddb(
                    creator=pymysql,
                    maxconnections=20,  # 最大连接数
                    mincached=2,  # 初始空闲连接
                    maxcached=10,  # 最大空闲连接
                    blocking=true,  # 连接耗尽时等待
                    ping=1,  # 使用时检查连接
                    **config
                )
        return cls._instance
            
    def get_connection(self):
        """从连接池获取连接"""
        return self._pool.connection()

# 使用连接池的数据库管理器
class pooleddbmanager:
    def __init__(self, pool_config: dict[str, any]):
        self.pool = connectionpool(pool_config)
    
    def execute_query(self, sql: str, params: tuple = none) -> list[dict]:
        """执行查询"""
        conn = self.pool.get_connection()
        try:
            with conn.cursor(dictcursor) as cursor:
                cursor.execute(sql, params or ())
                return cursor.fetchall()
        finally:
            conn.close()  # 实际是放回连接池
    
    def execute_update(self, sql: str, params: tuple = none) -> int:
        """执行更新"""
        conn = self.pool.get_connection()
        try:
            with conn.cursor() as cursor:
                affected_rows = cursor.execute(sql, params or ())
                conn.commit()
                return affected_rows
        except exception as e:
            conn.rollback()
            raise e
        finally:
            conn.close()

ping 参数说明

0 = 不检查
1 = 每次请求时检查（推荐）
2 = 每次游标创建时检查
4 = 每次执行时检查
7 = 1+2+4（所有检查）

五、应用示例

flask 集成示例

from dbutils.pooled_db import pooleddb
from flask import flask, request, jsonify
from pymysql.cursors import dictcursor

app = flask(__name__)

db_config = {
    'host': 'localhost',
    'user': 'root',
    'password': 'password',
    'database': 'test_db',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.dictcursor
}

# 初始化连接池
db_manager = pooleddbmanager(db_config)

@app.route('/users', methods=['get'])
def get_users():
    """获取所有用户"""
    try:
        users = db_manager.execute_query("select * from users")
        return jsonify({'success': true, 'data': users})
    except exception as e:
        return jsonify({'success': false, 'error': str(e)}), 500

@app.route('/users', methods=['post'])
def create_user():
    """创建用户"""
    try:
        data = request.json
        sql = "insert into users (name, email, age) values (%s, %s, %s)"
        result = db_manager.execute_update(sql, (data['name'], data['email'], data['age']))
        return jsonify({'success': true, 'affected_rows': result})
    except exception as e:
        return jsonify({'success': false, 'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=true)

连接池实践配置

# 优化后的连接池配置
optimal_pool_config = {
    'maxconnections': 20,      # 根据并发量调整
    'mincached': 2,           # 减少初始资源占用
    'maxcached': 10,          # 控制最大空闲连接
    'blocking': true,         # 避免连接耗尽错误
    'ping': 1,                # 使用前检查连接健康
    **db_config              # 基础数据库配置
}

错误重试机制

数据库操作重试装饰器：当数据库连接出现临时故障时，会自动进行最多3次重试，并且每次重试间隔时间按指数增长（1秒、2秒、4秒），提高程序的容错能力。

import time
from functools import wraps
import pymysql

def retry_on_failure(max_retries=3, initial_delay=1):
    """数据库操作重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (pymysql.operationalerror, pymysql.interfaceerror) as e:
                    if attempt == max_retries - 1:
                        raise e
                    time.sleep(initial_delay * (2 ** attempt))  # 指数退避
            return none
        return wrapper
    return decorator

# 使用示例
@retry_on_failure(max_retries=3)
def robust_query(sql, params=none):
    return db_manager.execute_query(sql, params)

指数退避：当操作失败时，不立即重试，而是等待一段时间，且每次重试的等待时间呈指数级增长。等待 1 秒， 2 秒， 4 秒，8 秒…

六、sql事务操作对比

事务影响

操作类型	语法示例	主要用途	返回值	事务影响	性能考虑	使用场景
select (查询)	select * from users where age > 18;	从数据库中检索数据	结果集（0行或多行）	只读操作，不影响数据	索引优化很重要，避免全表扫描	数据查询、报表生成、数据分析
update (更新)	update users set age = 20 where id = 1;	修改现有记录	受影响的行数	需要事务控制，会锁定行	where 条件要精确，避免锁表	修改用户信息、更新状态、调整数值
insert (插入)	insert into users (name, age) values (‘张三’, 25);	添加新记录	插入的行数（通常是1）	需要事务控制	批量插入比单条插入高效	新增用户、创建订单、记录日志
delete (删除)	delete from users where id = 1;	删除记录	受影响的行数	需要事务控制，谨慎使用	建议软删除，避免物理删除	删除用户、清理数据、撤销操作

事务特性

操作	是否自动提交	锁级别	回滚支持	并发影响
select	是（可设置）	共享锁	可回滚到快照	低（读写不阻塞）
update	否	排他锁	完全支持	高（会阻塞其他写操作）
insert	否	排他锁	完全支持	中（可能触发索引重建）
delete	否	排他锁	完全支持	高（会阻塞其他操作）

排他锁（x锁）：写锁，一个事务独占资源，其他事务不能读写
共享锁（s锁）：读锁，多个事务可同时读取，但不能写入
排他锁 = 独占，共享锁 = 共享读

普通 select 是完全无锁的，不会阻塞其他事务的写操作，也不会被写操作阻塞。只有显式加锁的select才会影响并发。

七、总结

连接管理

使用连接池管理数据库连接
合理配置连接池参数
及时释放连接回池

事务控制

明确控制事务边界
及时提交或回滚事务
处理并发场景下的数据一致性

错误处理

实现适当的重试机制
记录详细的错误日志
区分业务错误和系统错误

性能优化

使用预处理语句防止 sql 注入
合理使用批量操作
监控连接池使用情况

以上为个人经验，希望能给大家一个参考，也希望大家多多支持代码网。

PyMySQL数据库连接与优化方式

2025年10月09日 • Mysql •我要评论