当前位置： 首页 > news >正文

Python与MySQL高效集成指南：从基础到高级实践

news 2025/7/4 18:39:26

高级查询演示# Python与MySQL高效集成指南：从基础到高级实践

一、概述

Python与MySQL的结合是数据密集型应用开发中最常见的技术栈之一。本文全面介绍从基础连接到高级优化的各种技术，帮助开发者构建高效、可靠的数据库应用。

1.1 驱动选择对比

驱动名称类型特点适用场景
mysql-connector-python官方驱动纯Python实现，支持事务、预编译语句、连接池需要官方支持，对性能要求不苛刻
PyMySQL第三方驱动纯Python实现，API设计友好，兼容MySQLdb通用场景，轻量级应用
mysqlclient (MySQLdb)第三方驱动C扩展实现，速度最快高性能要求场景
aiomysql异步驱动基于PyMySQL的异步实现高并发、IO密集型应用

1.2 ORM框架对比

ORM名称特点适用场景
SQLAlchemy功能全面，高度灵活，支持原生SQL企业级应用，复杂查询需求
Peewee轻量级，语法简洁，易于学习中小型项目，API开发
Tortoise ORM专为异步设计，Django风格API异步应用，高并发服务
Django ORM与Django框架紧密集成Django Web应用

二、基础操作

2.1 驱动安装与选择策略

# 官方驱动
pip install mysql-connector-python# 社区驱动
pip install PyMySQL# C扩展版（性能更好）
pip install mysqlclient

驱动选择策略：

  • 对性能要求极高：优先选择mysqlclient
  • 需要异步支持：选择aiomysql或mysql.connector.aio
  • 无特殊要求：PyMySQL作为平衡性能和易用性的选择

2.2 建立连接与基本CRUD操作

PyMySQL示例
import pymysql
from pymysql.cursors import DictCursor# 建立连接
conn = pymysql.connect(host='127.0.0.1',port=3306,  # 默认端口，可省略user='root',password='password',database='testdb',charset='utf8mb4',  # 支持完整的UTF-8字符集，包括表情符号cursorclass=DictCursor  # 返回字典而非元组，方便字段访问
)try:# 创建游标with conn.cursor() as cursor:# 创建表cursor.execute("""CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(100) NOT NULL,email VARCHAR(100) UNIQUE,age INT,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""")# 插入数据cursor.execute("INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",('张三', 'zhangsan@example.com', 30))# 查询数据cursor.execute("SELECT * FROM users WHERE age > %s", (25,))result = cursor.fetchall()  # 获取所有结果for row in result:print(f"ID: {row['id']}, 姓名: {row['name']}, 邮箱: {row['email']}")# 更新数据cursor.execute("UPDATE users SET age = %s WHERE name = %s",(31, '张三'))# 删除数据cursor.execute("DELETE FROM users WHERE id = %s", (1,))# 提交更改（写操作必须）conn.commit()except Exception as e:# 发生错误时回滚conn.rollback()print(f"发生错误: {e}")finally:# 关闭连接conn.close()
mysql-connector-python示例
import mysql.connector# 建立连接
conn = mysql.connector.connect(host='127.0.0.1',user='root',password='password',database='testdb',auth_plugin='mysql_native_password'  # 明确指定认证插件
)try:# 创建游标cursor = conn.cursor(dictionary=True)  # 返回字典结果# 执行查询cursor.execute("SELECT * FROM users")# 获取单条结果first_user = cursor.fetchone()if first_user:print(f"第一条记录: {first_user}")# 获取指定数量结果next_three = cursor.fetchmany(3)print(f"接下来三条记录: {next_three}")# 获取所有剩余结果remaining = cursor.fetchall()print(f"剩余记录数量: {len(remaining)}")cursor.close()except mysql.connector.Error as err:print(f"MySQL错误: {err}")finally:if conn.is_connected():conn.close()

2.3 使用上下文管理器简化资源管理

import pymysql# 结合上下文管理器和异常处理
def execute_query(sql, params=None):with pymysql.connect(host='127.0.0.1', user='root', password='password', database='testdb', charset='utf8mb4') as conn:with conn.cursor() as cursor:try:cursor.execute(sql, params or ())if sql.strip().upper().startswith(('INSERT', 'UPDATE', 'DELETE')):conn.commit()return cursor.rowcount  # 返回影响的行数else:return cursor.fetchall()  # 返回查询结果except Exception as e:conn.rollback()raise e  # 重新抛出异常供上层处理# 使用示例
try:users = execute_query("SELECT * FROM users WHERE age BETWEEN %s AND %s", (18, 30))print(f"查询到{len(users)}个用户")
except Exception as e:print(f"查询失败: {e}")

三、连接池技术详解

在高并发环境中，频繁创建和销毁数据库连接会产生大量开销。连接池通过预先创建一组连接并复用它们来提高性能。

3.1 官方驱动连接池

from mysql.connector import pooling
import time# 创建连接池
pool = pooling.MySQLConnectionPool(pool_name="mypool",pool_size=5,  # 池中保持的连接数pool_reset_session=True,  # 重用前重置会话变量host='127.0.0.1',user='root',password='password',database='testdb'
)# 性能测试函数
def performance_test(query_count=1000):start_time = time.time()for i in range(query_count):try:# 从池中获取连接conn = pool.get_connection()cursor = conn.cursor()cursor.execute("SELECT 1")cursor.fetchall()cursor.close()conn.close()  # 归还到连接池，而不是真正关闭except Exception as e:print(f"查询 {i} 失败: {e}")duration = time.time() - start_timeprint(f"执行 {query_count} 次查询耗时: {duration:.2f}秒")print(f"平均每次查询: {(duration * 1000 / query_count):.2f}毫秒")# 执行测试
performance_test()

核心参数解释：

  • pool_size：连接池大小，通常设为预期并发连接的2-3倍
  • pool_reset_session：重用前是否重置会话状态
  • autocommit：连接默认自动提交状态
  • time_to_live：连接最大生存时间(ms)，超时后会被重新创建

3.2 DBUtils高级连接池配置

DBUtils库提供了两种连接池实现：

  • PersistentDB：为每个线程维护专用连接
  • PooledDB：通用连接池，支持多种复用策略
from DBUtils.PooledDB import PooledDB
import pymysql
import threading# 创建高级配置的连接池
pool = PooledDB(creator=pymysql,           # 使用的数据库模块maxconnections=10,         # 连接池允许的最大连接数mincached=2,               # 初始化时创建的空闲连接数maxcached=5,               # 连接池中允许的最大空闲连接数maxshared=3,               # 最大共享连接数blocking=True,             # 连接池满时是否阻塞等待maxusage=None,             # 单个连接的最大重用次数setsession=[],             # 可选的会话初始化命令列表ping=1,                    # 检查连接是否可用的方式(0=None,1=default,2=when_idle,4=when_accessed,7=always)host='127.0.0.1',user='root',password='password',database='testdb',charset='utf8mb4'
)# 线程安全的数据库操作示例
def thread_safe_query(thread_id):# 从连接池获取连接conn = pool.connection()cursor = conn.cursor()# 模拟查询操作cursor.execute("SELECT SLEEP(0.1)")  # 模拟耗时操作cursor.fetchall()# 归还连接到池cursor.close()conn.close()print(f"线程 {thread_id} 完成查询")# 创建多个线程并发执行查询
threads = []
for i in range(20):t = threading.Thread(target=thread_safe_query, args=(i,))threads.append(t)t.start()# 等待所有线程完成
for t in threads:t.join()

3.3 自定义连接池封装类

创建一个可重用的连接池管理类：

import pymysql
from DBUtils.PooledDB import PooledDB
import threadingclass MySQLPool:"""MySQL连接池管理类"""_instance_lock = threading.Lock()# 单例模式实现def __new__(cls, *args, **kwargs):if not hasattr(cls, "_instance"):with cls._instance_lock:if not hasattr(cls, "_instance"):cls._instance = object.__new__(cls)return cls._instancedef __init__(self, config=None):if hasattr(self, "_pool"):return# 默认配置self.config = {'host': '127.0.0.1','port': 3306,'user': 'root','password': 'password','database': 'testdb','charset': 'utf8mb4','maxconnections': 10,'mincached': 3,'maxcached': 5,'blocking': True,'ping': 1}# 更新配置if config:self.config.update(config)# 创建连接池self._pool = PooledDB(creator=pymysql,**self.config)def get_conn(self):"""获取数据库连接"""return self._pool.connection()def execute_query(self, sql, params=None, fetch=True):"""执行查询SQL"""conn = self.get_conn()cursor = conn.cursor(pymysql.cursors.DictCursor)result = Nonetry:cursor.execute(sql, params or ())conn.commit()if fetch:result = cursor.fetchall()else:result = cursor.rowcountexcept Exception as e:conn.rollback()raise efinally:cursor.close()conn.close()return resultdef execute_many(self, sql, params_list):"""批量执行SQL"""conn = self.get_conn()cursor = conn.cursor()row_count = 0try:row_count = cursor.executemany(sql, params_list)conn.commit()except Exception as e:conn.rollback()raise efinally:cursor.close()conn.close()return row_count# 使用示例
db_pool = MySQLPool({'database': 'myproject','maxconnections': 20
})# 执行查询
users = db_pool.execute_query("SELECT * FROM users WHERE status = %s", ('active',))
print(f"活跃用户数量: {len(users)}")# 批量插入
data = [(f'user{i}', f'user{i}@example.com') for i in range(1, 101)]
affected = db_pool.execute_many("INSERT INTO users (username, email) VALUES (%s, %s)", data
)
print(f"插入记录数: {affected}")

3.4 连接池监控与管理

监控连接池的健康状态和性能是生产环境中的关键任务：

class PoolMonitor:"""连接池监控器"""def __init__(self, pool):self.pool = poolself._monitor_data = {'created_connections': 0,'reused_connections': 0,'active_connections': 0,'max_active_connections': 0,'total_wait_time': 0,'queries_executed': 0}self._lock = threading.Lock()def connection_created(self):with self._lock:self._monitor_data['created_connections'] += 1def connection_reused(self):with self._lock:self._monitor_data['reused_connections'] += 1def connection_active(self):with self._lock:self._monitor_data['active_connections'] += 1self._monitor_data['max_active_connections'] = max(self._monitor_data['max_active_connections'],self._monitor_data['active_connections'])def connection_released(self):with self._lock:self._monitor_data['active_connections'] -= 1def query_executed(self, wait_time):with self._lock:self._monitor_data['queries_executed'] += 1self._monitor_data['total_wait_time'] += wait_timedef get_stats(self):with self._lock:stats = self._monitor_data.copy()# 计算平均等待时间if stats['queries_executed'] > 0:stats['avg_wait_time'] = stats['total_wait_time'] / stats['queries_executed']else:stats['avg_wait_time'] = 0return statsdef reset(self):with self._lock:for key in self._monitor_data:self._monitor_data[key] = 0# 使用示例将在实战部分展示

四、大容量与批量操作技术

处理大量数据时，性能优化至关重要。本节详细介绍如何高效执行批量操作和处理大规模数据集。

4.1 批量插入对比与优化

基本批量插入方法对比
import pymysql
import timeconn = pymysql.connect(host='127.0.0.1', user='root', password='password', database='testdb')
cursor = conn.cursor()# 准备测试数据 - 10,000条记录
data = [(i, f'user{i}', f'user{i}@example.com', 20 + (i % 30)) for i in range(1, 10001)]# 方法1: 单条插入
def single_insert(data):start_time = time.time()for item in data:cursor.execute("INSERT INTO users (id, name, email, age) VALUES (%s, %s, %s, %s)",item)conn.commit()return time.time() - start_time# 方法2: executemany批量插入
def executemany_insert(data):start_time = time.time()cursor.executemany("INSERT INTO users (id, name, email, age) VALUES (%s, %s, %s, %s)",data)conn.commit()return time.time() - start_time# 方法3: 拼接多值语句(注意防SQL注入)
def multi_value_insert(data):start_time = time.time()# 构造INSERT INTO users VALUES (1,'a','a@a',20),(2,'b','b@b',21),...形式的SQLchunks = [data[i:i+1000] for i in range(0, len(data), 1000)]for chunk in chunks:placeholder = ','.join(['(%s,%s,%s,%s)'] * len(chunk))flat_data = [item for sublist in chunk for item in sublist]sql = f"INSERT INTO users (id, name, email, age) VALUES {placeholder}"cursor.execute(sql, flat_data)conn.commit()return time.time() - start_time# 测试并打印结果
print(f"单条插入耗时: {single_insert(data[:100]):.2f}秒")  # 只测试100条避免太慢
print(f"executemany批量插入耗时: {executemany_insert(data):.2f}秒") 
print(f"多值插入耗时: {multi_value_insert(data):.2f}秒")cursor.close()
conn.close()

性能对比分析：

  • 单条插入：最慢，每条记录一次网络往返和提交
  • executemany：中等，减少了Python到MySQL的往返次数
  • 多值语句：最快，一次网络往返插入多条记录，减少解析开销
优化批量插入的关键参数
import pymysqlconn = pymysql.connect(host='127.0.0.1', user='root',password='password', database='testdb'
)# 重要: 设置自动提交为False提升批量性能
conn.autocommit = Falsetry:cursor = conn.cursor()# 准备批量数据data = [(i, f'name{i}') for i in range(1, 100001)]# 1. 调整MySQL会话变量以优化插入性能cursor.execute("SET foreign_key_checks = 0")  # 暂时关闭外键检查cursor.execute("SET unique_checks = 0")       # 暂时关闭唯一性检查cursor.execute("SET sql_log_bin = 0")         # 暂时关闭二进制日志# 2. 分批处理，避免单次传输过多数据batch_size = 5000for i in range(0, len(data), batch_size):batch = data[i:i+batch_size]cursor.executemany("INSERT INTO users (id, name) VALUES (%s, %s)",batch)# 每批次手动提交，平衡提交频率和事务大小conn.commit()# 3. 恢复MySQL会话变量cursor.execute("SET foreign_key_checks = 1")cursor.execute("SET unique_checks = 1")cursor.execute("SET sql_log_bin = 1")except Exception as e:conn.rollback()print(f"批量插入失败: {e}")finally:cursor.close()conn.close()

4.2 高效导入外部数据文件

LOAD DATA INFILE高速导入
import pymysql
import csv
import os# 1. 准备CSV数据文件
def prepare_csv_file(filename, records):with open(filename, 'w', newline='') as f:writer = csv.writer(f)for record in records:writer.writerow(record)return os.path.abspath(filename)# 生成示例数据并创建CSV
data = [(i, f'user{i}', f'user{i}@example.com', 20 + (i % 30)) for i in range(1, 1000001)]  # 一百万条记录
csv_path = prepare_csv_file('users_data.csv', data)# 2. 使用LOAD DATA INFILE导入
conn = pymysql.connect(host='127.0.0.1', user='root', password='password', database='testdb',local_infile=True)  # 必须启用local_infile选项
cursor = conn.cursor()try:# 必须启用MySQL服务器的local_infile选项才能使用本地文件# 在my.cnf配置中设置: local_infile=1# 或在会话中设置: SET GLOBAL local_infile=1;# 执行LOAD DATA命令(注意替换路径分隔符为Unix风格)load_data_sql = f"""LOAD DATA LOCAL INFILE '{csv_path.replace('\\', '/')}'INTO TABLE usersFIELDS TERMINATED BY ',' LINES TERMINATED BY '\\n'(id, name, email, age)"""cursor.execute(load_data_sql)conn.commit()print(f"已导入{cursor.rowcount}条记录")except Exception as e:conn.rollback()print(f"数据导入失败: {e}")finally:cursor.close()conn.close()

LOAD DATA优化提示：

  • 确保服务器启用了local_infile选项
  • 对于超大文件，考虑暂时禁用索引，导入后再重建
  • 使用FIELDS ENCLOSED BY处理包含分隔符的字段
  • 配合SET子句可在导入时进行数据转换
mysqlimport命令行工具集成
import subprocess
import os
import csv
import time# 准备数据文件
def generate_large_csv(filename, rows=1000000):with open(filename, 'w', newline='') as f:writer = csv.writer(f)for i in range(rows):writer.writerow([i, f'user{i}', f'user{i}@example.com', 25])return os.path.abspath(filename)# 使用mysqlimport导入
def mysql_import_csv(csv_path, table_name, db_config):start_time = time.time()# 构建mysqlimport命令cmd = ['mysqlimport','--local',f'--host={db_config["host"]}',f'--user={db_config["user"]}',f'--password={db_config["password"]}','--fields-terminated-by=,','--fields-optionally-enclosed-by="','--lines-terminated-by=\\n',db_config["database"],csv_path  # 文件名必须与表名匹配]# 执行导入命令result = subprocess.run(cmd, capture_output=True, text=True)duration = time.time() - start_timeif result.returncode == 0:print(f"数据导入成功，耗时: {duration:.2f}秒")print(result.stdout)else:print(f"数据导入失败: {result.stderr}")return result.returncode == 0, duration# 使用示例
config = {"host": "127.0.0.1","user": "root","password": "password","database": "testdb"
}# 注意: 生成的CSV文件名必须与数据库表名匹配
csv_file = generate_large_csv('users.csv')
mysql_import_csv(csv_file, 'users', config)

4.3 流式处理大结果集

处理大数据集时，一次性加载所有结果可能导致内存溢出。以下是几种流式处理方法：

使用SSCursor逐行处理
import pymysql
import pymysql.cursorsconn = pymysql.connect(host='127.0.0.1',user='root',password='password',database='testdb',cursorclass=pymysql.cursors.SSCursor  # 流式游标
)try:with conn.cursor() as cursor:# 查询可能返回大量结果的表cursor.execute("SELECT * FROM large_table")# 逐行处理结果，避免一次性加载到内存row_count = 0while True:row = cursor.fetchone()if row is None:break# 处理单行数据process_row(row)row_count += 1# 每处理1000行打印一次进度if row_count % 1000 == 0:print(f"已处理 {row_count} 行...")print(f"总共处理了 {row_count} 行数据")except Exception as e:print(f"处理数据时出错: {e}")finally:conn.close()def process_row(row):"""处理单行数据的函数"""# 这里是对每行数据的处理逻辑pass
分页查询处理超大表
import pymysqlconn = pymysql.connect(host='127.0.0.1',user='root',password='password',database='testdb'
)try:cursor = conn.cursor(pymysql.cursors.DictCursor)# 分页参数page_size = 1000last_id = 0total_processed = 0while True:# 使用主键分页查询，比OFFSET效率更高cursor.execute("SELECT * FROM large_table WHERE id > %s ORDER BY id LIMIT %s",(last_id, page_size))# 获取本页数据rows = cursor.fetchall()# 检查是否还有数据if not rows:break# 处理本页数据for row in rows:process_data(row)last_id = row['id']  # 更新最后处理的IDtotal_processed += len(rows)print(f"已处理 {total_processed} 条记录...")print(f"总共处理记录: {total_processed}")except Exception as e:print(f"分页查询出错: {e}")finally:cursor.close()conn.close()def process_data(row):"""处理数据的函数"""# 对数据的处理逻辑pass

分页查询性能提示：

  • 避免使用OFFSET，对大表性能极差
  • 使用"断点式"查询（基于上次查询的最后ID）
  • 确保分页字段有索引
  • 根据数据量和内存调整批量大小

4.4 高级技巧：并行批处理

利用Python多线程或多进程并行处理数据：

import pymysql
import threading
import queue
import time
from concurrent.futures import ThreadPoolExecutor# 初始化数据库连接池
from DBUtils.PooledDB import PooledDBpool = PooledDB(creator=pymysql,maxconnections=10,mincached=5,host='127.0.0.1',user='root',password='password',database='testdb'
)# 任务队列
task_queue = queue.Queue()# 准备任务：将ID范围划分为多个小批次
def prepare_tasks(total_records, batch_size=1000):for start_id in range(1, total_records + 1, batch_size):end_id = min(start_id + batch_size - 1, total_records)task_queue.put((start_id, end_id))# 工作线程函数
def worker():# 获取线程专用连接conn = pool.connection()cursor = conn.cursor()while not task_queue.empty():try:# 获取下一个任务批次start_id, end_id = task_queue.get(block=False)# 处理该批次数据cursor.execute("SELECT * FROM large_table WHERE id BETWEEN %s AND %s",(start_id, end_id))batch_data = cursor.fetchall()# 进行批量处理process_batch(cursor, batch_data)conn.commit()print(f"处理完成ID范围: {start_id}-{end_id}")except queue.Empty:breakexcept Exception as e:conn.rollback()print(f"处理批次出错: {e}")finally:task_queue.task_done()# 释放资源cursor.close()conn.close()# 处理一批数据的函数
def process_batch(cursor, batch_data):# 批量处理逻辑，例如批量更新update_data = []for row in batch_data:# 对每行数据进行处理，准备更新数据processed_value = process_row(row)update_data.append((processed_value, row[0]))  # 假设第一列是ID# 批量更新if update_data:cursor.executemany("UPDATE large_table SET processed_value = %s WHERE id = %s",update_data)# 数据处理函数示例
def process_row(row):# 模拟复杂处理time.sleep(0.001)return f"Processed-{row[0]}"# 主函数：启动并行处理
def parallel_process(total_records, num_workers=4, batch_size=1000):start_time = time.time()# 准备任务prepare_tasks(total_records, batch_size)# 创建工作线程池with ThreadPoolExecutor(max_workers=num_workers) as executor:for _ in range(num_workers):executor.submit(worker)duration = time.time() - start_timeprint(f"并行处理 {total_records} 条记录完成")print(f"总耗时: {duration:.2f}秒，平均每秒处理 {total_records/duration:.2f} 条记录")# 执行并行批处理
parallel_process(100000, num_workers=4, batch_size=5000)

五、ORM集成：从原生SQL到对象映射

ORM（对象关系映射）是一种编程技术，将关系型数据库中的数据转换为面向对象编程语言中的对象。Python中有多种成熟的ORM框架，本节将详细介绍几种主流ORM的集成与使用。

5.1 SQLAlchemy：功能全面的ORM框架

SQLAlchemy是Python最流行的ORM框架之一，具备高度灵活性和强大的功能。

安装与基本配置
# 安装
# pip install sqlalchemy pymysqlfrom sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
import datetime# 创建数据库引擎
engine = create_engine('mysql+pymysql://root:password@localhost/testdb',echo=False,  # 设为True可以显示生成的SQL语句pool_size=5,pool_recycle=3600,  # 连接超过1小时后回收pool_pre_ping=True  # 每次使用前测试连接是否可用
)# 创建基类
Base = declarative_base()# 定义模型类
class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True, autoincrement=True)username = Column(String(50), nullable=False, unique=True)email = Column(String(100), nullable=False)created_at = Column(DateTime, default=datetime.datetime.now)# 关联关系posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")def __repr__(self):return f"<User(username='{self.username}', email='{self.email}')>"class Post(Base):__tablename__ = 'posts'id = Column(Integer, primary_key=True, autoincrement=True)title = Column(String(100), nullable=False)content = Column(String(1000))user_id = Column(Integer, ForeignKey('users.id'), nullable=False)# 关联关系author = relationship("User", back_populates="posts")def __repr__(self):return f"<Post(title='{self.title}')>"# 创建表结构
Base.metadata.create_all(engine)# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
基本CRUD操作
# 创建用户
new_user = User(username='alice', email='alice@example.com')
session.add(new_user)
session.commit()# 批量添加
users = [User(username='bob', email='bob@example.com'),User(username='charlie', email='charlie@example.com')
]
session.add_all(users)
session.commit()# 查询单个用户
user = session.query(User).filter_by(username='alice').first()
print(user)# 复杂查询
from sqlalchemy import or_, and_, desc
users = session.query(User).filter(or_(User.username.like('a%'),User.email.contains('example'))
).order_by(desc(User.created_at)).all()for user in users:print(f"Username: {user.username}, Email: {user.email}")# 更新数据
user_to_update = session.query(User).filter_by(username='bob').first()
if user_to_update:user_to_update.email = 'bob_new@example.com'session.commit()# 删除数据
user_to_delete = session.query(User).filter_by(username='charlie').first()
if user_to_delete:session.delete(user_to_delete)session.commit()
事务与异常处理
# 使用事务
try:# 开始事务user = User(username='dave', email='dave@example.com')session.add(user)post = Post(title='First Post', content='Hello World', user_id=user.id)session.add(post)# 如果一切正常，提交事务session.commit()
except Exception as e:# 发生异常，回滚事务session.rollback()print(f"错误: {e}")
finally:session.close()
高级查询技巧
# 连接查询
from sqlalchemy.orm import joinedload# 预加载关联对象（避免N+1查询问题）
users_with_posts = session.query(User).options(joinedload(User.posts)).all()for user in users_with_posts:print(f"User: {user.username}")for post in user.posts:  # 不会产生额外的查询print(f"  - Post: {post.title}")# 聚合查询
from sqlalchemy import func# 计算每个用户的帖子数量
post_counts = session.query(User.username, func.count(Post.id).label('post_count')
).join(Post).group_by(User.id).all()for username, count in post_counts:print(f"User {username} has {count} posts")# 原生SQL
result = session.execute(text("SELECT u.username, COUNT(p.id) as post_count FROM users u JOIN posts p ON u.id = p.user_id GROUP BY u.id")
)
for row in result:print(f"User {row.username} has {row.post_count} posts")
数据库迁移与版本控制

SQLAlchemy通常与Alembic一起使用来管理数据库架构变更：

# 安装Alembic
pip install alembic# 初始化Alembic环境
alembic init migrations# 创建迁移脚本
alembic revision --autogenerate -m "Create users and posts tables"# 执行迁移
alembic upgrade head# 回滚迁移
alembic downgrade -1

5.2 Peewee：轻量级ORM框架

Peewee是一个简单而强大的ORM，适合中小型项目，API设计简洁易用。

基本模型定义与操作
# 安装
# pip install peeweefrom peewee import *
import datetime# 定义数据库连接
db = MySQLDatabase('testdb', user='root', password='password',host='localhost', port=3306,charset='utf8mb4'
)# 定义基础模型类
class BaseModel(Model):class Meta:database = db# 定义用户模型
class User(BaseModel):username = CharField(max_length=50, unique=True)email = CharField(max_length=100)created_at = DateTimeField(default=datetime.datetime.now)class Meta:table_name = 'users'# 定义帖子模型
class Post(BaseModel):title = CharField(max_length=100)content = TextField(null=True)user = ForeignKeyField(User, backref='posts')created_at = DateTimeField(default=datetime.datetime.now)class Meta:table_name = 'posts'# 创建表
db.connect()
db.create_tables([User, Post])# 插入数据
with db.atomic():user = User.create(username='john', email='john@example.com')Post.create(title='Hello Peewee', content='My first post', user=user)# 查询数据
users = User.select()
for user in users:print(f"User: {user.username}, Email: {user.email}")# 更新数据
query = User.update(email='john_updated@example.com').where(User.username == 'john')
query.execute()# 删除数据
query = Post.delete().where(Post.title == 'Hello Peewee')
query.execute()# 关闭连接
db.close()

5.3 Tortoise ORM：异步ORM框架

Tortoise ORM是为异步IO设计的现代ORM框架，配合FastAPI等异步框架特别合适。

基本使用示例
# 安装
# pip install tortoise-orm aiomysqlfrom tortoise import Tortoise, fields, run_async
from tortoise.models import Model# 定义模型
class User(Model):id = fields.IntField(pk=True)username = fields.CharField(max_length=50, unique=True)email = fields.CharField(max_length=100)created_at = fields.DatetimeField(auto_now_add=True)# 反向关系posts: fields.ReverseRelation["Post"]class Meta:table = "users"class Post(Model):id = fields.IntField(pk=True)title = fields.CharField(max_length=100)content = fields.TextField(null=True)user = fields.ForeignKeyField("models.User", related_name="posts")created_at = fields.DatetimeField(auto_now_add=True)class Meta:table = "posts"# 异步初始化函数
async def init():# 初始化ORMawait Tortoise.init(db_url='mysql://root:password@localhost/testdb',modules={'models': ['__main__']}  # 当前文件中的模型)# 创建表await Tortoise.generate_schemas()# 异步操作示例
async def run():await init()# 创建用户user = await User.create(username='async_user', email='async@example.com')# 创建帖子await Post.create(title='Async Post', content='Using Tortoise ORM', user=user)# 查询用户及其帖子users = await User.all().prefetch_related('posts')for user in users:print(f"User: {user.username}")for post in user.posts:print(f"  - Post: {post.title}")# 关闭连接await Tortoise.close_connections()# 运行异步代码
if __name__ == "__main__":run_async(run())

5.4 Django ORM：完整Web框架的一部分

如果你正在使用Django框架，它内置了功能强大的ORM系统。

Django ORM基本使用
# 在Django项目中的models.py文件
from django.db import modelsclass User(models.Model):username = models.CharField(max_length=50, unique=True)email = models.EmailField()created_at = models.DateTimeField(auto_now_add=True)def __str__(self):return self.usernameclass Post(models.Model):title = models.CharField(max_length=100)content = models.TextField(blank=True)user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')created_at = models.DateTimeField(auto_now_add=True)def __str__(self):return self.title# 在Django视图或其他地方使用
def view_example(request):# 创建用户user = User.objects.create(username='django_user', email='django@example.com')# 创建帖子Post.objects.create(title='Django Post', content='Using Django ORM', user=user)# 查询用户及帖子users = User.objects.all().prefetch_related('posts')for user in users:posts = user.posts.all()# 处理数据# 复杂查询from django.db.models import Count, Qactive_users = User.objects.annotate(post_count=Count('posts')).filter(post_count__gt=0)# 返回数据return render(request, 'template.html', {'users': users})

六、异步操作

在现代高并发应用中，异步IO是提升性能和资源利用率的关键技术。本节介绍Python中操作MySQL的几种主要异步方案。

6.1 官方异步驱动：mysql.connector.aio

MySQL官方连接器提供了异步API支持，适合与asyncio集成：

import asyncio
from mysql.connector import aioasync def main():# 建立异步连接conn = await aio.connect(host='127.0.0.1',user='root',password='password',database='testdb')# 创建游标cursor = await conn.cursor()# 执行查询await cursor.execute("SELECT id, name, email FROM users LIMIT 10")# 获取结果rows = await cursor.fetchall()for row in rows:print(f"User: {row[1]}, Email: {row[2]}")# 关闭资源await cursor.close()await conn.close()# 运行异步函数
asyncio.run(main())

6.2 第三方异步驱动：aiomysql

aiomysql是基于PyMySQL的异步版本，提供完整的异步API：

import asyncio
import aiomysqlasync def main():# 创建连接池pool = await aiomysql.create_pool(host='127.0.0.1',user='root',password='password',db='testdb',charset='utf8mb4',minsize=1,maxsize=10,autocommit=False)async with pool.acquire() as conn:# 执行读取操作async with conn.cursor(aiomysql.DictCursor) as cursor:await cursor.execute("SELECT * FROM users WHERE age > %s", (25,))result = await cursor.fetchall()for row in result:print(f"ID: {row['id']}, Name: {row['name']}")# 执行写入操作(带事务)async with conn.cursor() as cursor:try:await cursor.execute("INSERT INTO logs (message) VALUES (%s)", ("Async operation",))await conn.commit()except Exception as e:await conn.rollback()print(f"Error: {e}")# 关闭连接池pool.close()await pool.wait_closed()# 运行异步函数
asyncio.run(main())

6.3 异步ORM：SQLAlchemy 2.0

SQLAlchemy 2.0提供了完全异步的API，结合asyncio实现高效的数据库操作：

# 安装
# pip install sqlalchemy[asyncio] aiomysqlimport asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.future import select
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker
import datetimeBase = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True, autoincrement=True)username = Column(String(50), nullable=False)email = Column(String(100))created_at = Column(DateTime, default=datetime.datetime.now)async def main():# 创建异步引擎engine = create_async_engine('mysql+aiomysql://root:password@localhost/testdb',echo=False,)# 创建会话async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)# 创建表(仅首次运行需要)async with engine.begin() as conn:await conn.run_sync(Base.metadata.create_all)# 使用会话执行操作async with async_session() as session:# 插入数据new_user = User(username='async_sqlalchemy', email='async@example.com')session.add(new_user)await session.commit()# 查询数据result = await session.execute(select(User).where(User.username.like('async%')))users = result.scalars().all()for user in users:print(f"Username: {user.username}, Email: {user.email}")# 关闭引擎await engine.dispose()# 运行异步函数
asyncio.run(main())

6.4 高并发场景下的最佳实践

异步使用注意事项

  1. 避免阻塞异步循环：

    # 错误示例 - 会阻塞整个事件循环
async def process_data():async with pool.acquire() as conn:await cursor.execute("SELECT * FROM large_table")result = await cursor.fetchall()  # 大结果集会阻塞process_large_data(result)  # CPU密集操作会阻塞事件循环# 正确示例 - 使用流式处理和线程池
async def process_data():async with pool.acquire() as conn:await cursor.execute("SELECT * FROM large_table")# 逐条处理结果while True:row = await cursor.fetchone()if row is None:break# 将CPU密集任务放入线程池await asyncio.to_thread(process_row, row)

  2. 连接池管理：

    # 创建全局连接池
async def init_db():global poolpool = await aiomysql.create_pool(host='127.0.0.1', user='root', password='password',db='testdb', minsize=5, maxsize=20,autocommit=False, echo=True)# 应用启动时初始化
async def startup():await init_db()# 应用关闭时清理
async def shutdown():if pool:pool.close()await pool.wait_closed()

  3. 并发控制：

    # 限制并发查询数量
async def process_batch(batch_ids):semaphore = asyncio.Semaphore(10)  # 最多10个并发查询tasks = []async def process_id(id):async with semaphore:async with pool.acquire() as conn:async with conn.cursor() as cursor:await cursor.execute("SELECT * FROM users WHERE id = %s", (id,))result = await cursor.fetchone()return process_result(result)# 创建所有任务for id in batch_ids:tasks.append(asyncio.create_task(process_id(id)))# 等待所有任务完成results = await asyncio.gather(*tasks)return results

七、性能优化与最佳实践总结

将前述章节的关键优化点和最佳实践汇总如下：

7.1 连接管理优化

  1. 使用连接池：

    • 避免频繁创建/销毁连接
    • 设置合理的池大小（通常为预期并发量的2-3倍）
    • 定期测试连接活性（ping）

  2. 连接参数优化：

    conn = pymysql.connect(host='127.0.0.1',user='root',password='password',database='testdb',charset='utf8mb4',connect_timeout=10,        # 连接超时时间read_timeout=30,           # 读取超时时间write_timeout=30,          # 写入超时时间autocommit=False,          # 批量操作时关闭自动提交ssl={'ca': '/path/to/ca'}  # 生产环境启用SSL
)

  3. 连接错误处理：

    max_retries = 3
retry_count = 0while retry_count < max_retries:try:conn = pool.get_connection()# 操作数据库breakexcept mysql.connector.Error as err:if err.errno == errorcode.CR_SERVER_GONE_ERROR or err.errno == errorcode.CR_SERVER_LOST:retry_count += 1time.sleep(0.5)  # 指数退避策略更佳continueelse:raise

7.2 查询优化

  1. 预编译语句(Prepared Statements)：

    # MySQL Connector支持预编译
cursor = conn.cursor(prepared=True)
stmt = "SELECT * FROM users WHERE age > %s AND status = %s"
cursor.execute(stmt, (18, 'active'))

  2. 批量操作优化：

    • 使用executemany()而非循环单条执行
    • 考虑多值插入语法以减少网络往返
    • 大批量导入时使用LOAD DATA INFILE

  3. 避免N+1查询问题：

    # 不佳：会导致N+1查询
users = session.query(User).all()  # 1次查询
for user in users:posts = user.posts.all()       # N次查询# 优化：预加载关联对象
users = session.query(User).options(joinedload(User.posts)).all()  # 1次查询

  4. 查询字段精简：

    # 不佳：查询不需要的字段
cursor.execute("SELECT * FROM large_table")# 优化：只查询需要的字段
cursor.execute("SELECT id, name, email FROM large_table")

7.3 事务与锁优化

  1. 合理使用事务：

    # 批量操作放在同一事务中
conn.autocommit = False
cursor.execute("BEGIN")
try:for i in range(1000):cursor.execute("INSERT INTO logs VALUES (%s)", (f"Log {i}",))conn.commit()
except:conn.rollback()

  2. 避免长事务：

    • 对大量数据操作，分割为多个小事务
    • 使用合理的批量大小（5000-10000）平衡性能和内存使用

  3. 锁的考量：

    # 在会话级别设置隔离级别
cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")# 使用行锁而非表锁
cursor.execute("SELECT * FROM users WHERE id = 1 FOR UPDATE")

7.4 服务端优化

  1. 会话优化：

    # 批量插入前禁用外键检查
cursor.execute("SET foreign_key_checks = 0")# 批量插入前调整缓冲区
cursor.execute("SET unique_checks = 0")
cursor.execute("SET autocommit = 0")# 完成后恢复设置
cursor.execute("SET foreign_key_checks = 1")
cursor.execute("SET unique_checks = 1")
cursor.execute("SET autocommit = 1")

  2. 合理使用索引：

    • 常用查询条件字段创建索引
    • 避免过多索引影响写入性能
    • 利用EXPLAIN分析查询性能

7.5 应用层设计优化

  1. 数据访问层封装：

    class UserDAO:def __init__(self, db_pool):self.db_pool = db_pooldef get_by_id(self, user_id):conn = self.db_pool.get_conn()try:cursor = conn.cursor(pymysql.cursors.DictCursor)cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))return cursor.fetchone()finally:conn.close()# 其他CRUD方法...

  2. 缓存策略：

    import redis# 缓存结合数据库查询
def get_user(user_id):# 尝试从缓存获取cache_key = f"user:{user_id}"cached_user = redis_client.get(cache_key)if cached_user:return json.loads(cached_user)# 缓存未命中，从数据库查询user = db.execute_query("SELECT * FROM users WHERE id = %s", (user_id,))if user:# 更新缓存，设置过期时间redis_client.setex(cache_key, 3600, json.dumps(user))return user

  3. 读写分离：

    # 配置主从连接池
master_pool = create_pool(host='master.db', ...)
slave_pool = create_pool(host='slave.db', ...)def execute_query(sql, params=None):"""执行只读查询，使用从库"""if sql.strip().upper().startswith(('SELECT')):conn = slave_pool.get_connection()else:conn = master_pool.get_connection()try:# 执行查询...passfinally:conn.close()

八、总结

本文全面介绍了Python操作MySQL的各种技术和优化手段，从基础连高级查询演示接到高级特性，包括：

  1. 驱动与ORM选择：根据应用需求选择合适的驱动和ORM框架
  2. 连接池技术：有效管理数据库连接资源
  3. 批量操作优化：高效处理大量数据
  4. 异步编程：提升高并发场景下的性能
  5. 性能优化最佳实践：从多个层面提升应用性能

在这里插入图片描述

http://www.xdnf.cn/news/246727.html

相关文章：

  • Hibernate与MybatisPlus的混用问题(Invalid bound statement (not found))
  • （C题｜社交媒体平台用户分析问题）2025年第二十二届五一数学建模竞赛（五一杯/五一赛）解题思路|完整代码论文集合
  • 恒流源电路
  • RAG工程-基于LangChain 实现 Advanced RAG（预检索-查询优化）（下）
  • 前端HTML基础知识
  • 【大模型面试每日一题】Day 5：GQA vs MHA效率对比
  • CV中常用Backbone-2：ConvNeXt模型详解
  • 微软推出数款Phi 4“开放式”人工智能模型
  • VB.net序列化和反序列化的使用方法和实用场景
  • NUS：多模态多视角理解评估
  • 攻防世界 - Misc - Level 6 | Wireshark
  • jupyterlab建议安装的两个插件
  • LeetCode：DP-回文串问题
  • 如何测试登录模块？全面测试思路解析
  • Python爬虫（14）Python爬虫数据存储新范式：云原生NoSQL服务实战与运维成本革命
  • Socket通信
  • Beetle-RP2350 扩展板设计
  • 力扣——23合并升序链表
  • 【ESP32】st7735s + LVGL使用-------图片显示
  • python多线程输入字符和写入文件
  • 关系型数据库设计指南
  • 2025五一杯数学建模竞赛选题建议+初步分析
  • terraform实现本地加密与解密
  • sftp连接报错Received message too long 168449893
  • 大鱼吃小鱼开源
  • leetcode 977. Squares of a Sorted Array
  • 【免费】1992-2021年各省GDP数据/各省地区生产总值数据
  • GoogleTest：简单示例及ASSERT/EXPECT说明
  • [FPGA 官方 IP] Binary Counter
  • 多节点监测任务分配方法比较与分析