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

Python操作MySQL：从基础连接到CRUD实战

news 2026/7/16 11:53:53

1. Python操作MySQL基础环境搭建

1.1 MySQL驱动选择与安装

Python连接MySQL数据库需要依赖特定的数据库驱动。目前主流的选择有：

  • mysql-connector-python（MySQL官方驱动）
  • PyMySQL（纯Python实现）
  • MySQLdb（C扩展实现）

对于新手而言，我推荐使用mysql-connector-python这个官方驱动。安装方法很简单，使用pip命令即可：

pip install mysql-connector-python

注意：如果同时安装了多个Python版本，请确保pip命令对应的是你当前使用的Python版本。可以通过python -m pip install来明确指定。

1.2 数据库连接配置

建立数据库连接是操作MySQL的第一步，需要准备以下信息：

  • 主机地址（通常是localhost）
  • 用户名和密码
  • 数据库名称（可选）
import mysql.connector config = { 'host': 'localhost', 'user': 'your_username', 'password': 'your_password', 'database': 'your_database', # 可选 'port': 3306, # 默认端口 'charset': 'utf8mb4' # 推荐字符集 } try: conn = mysql.connector.connect(**config) print("数据库连接成功！") except mysql.connector.Error as err: print(f"连接失败: {err}") finally: if 'conn' in locals() and conn.is_connected(): conn.close()

实操心得：在生产环境中，建议将数据库配置信息存储在环境变量或配置文件中，不要直接硬编码在代码里。

2. CRUD基础操作实战

2.1 创建数据表

我们先创建一个简单的用户表作为示例：

def create_users_table(conn): cursor = conn.cursor() create_table_sql = """ CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, email VARCHAR(100) NOT NULL UNIQUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, is_active BOOLEAN DEFAULT TRUE ) """ try: cursor.execute(create_table_sql) conn.commit() print("用户表创建成功") except mysql.connector.Error as err: print(f"创建表失败: {err}") finally: cursor.close()

2.2 插入数据

单条插入和批量插入的示例：

def insert_user(conn, username, email): cursor = conn.cursor() insert_sql = "INSERT INTO users (username, email) VALUES (%s, %s)" try: cursor.execute(insert_sql, (username, email)) conn.commit() print(f"用户 {username} 插入成功，ID: {cursor.lastrowid}") except mysql.connector.Error as err: print(f"插入失败: {err}") finally: cursor.close() def batch_insert_users(conn, user_list): cursor = conn.cursor() insert_sql = "INSERT INTO users (username, email) VALUES (%s, %s)" try: cursor.executemany(insert_sql, user_list) conn.commit() print(f"批量插入了 {cursor.rowcount} 条记录") except mysql.connector.Error as err: print(f"批量插入失败: {err}") finally: cursor.close()

2.3 查询数据

基础查询和参数化查询示例：

def get_all_users(conn): cursor = conn.cursor(dictionary=True) # 返回字典形式的结果 select_sql = "SELECT * FROM users" try: cursor.execute(select_sql) return cursor.fetchall() except mysql.connector.Error as err: print(f"查询失败: {err}") return [] finally: cursor.close() def search_users(conn, keyword): cursor = conn.cursor(dictionary=True) search_sql = """ SELECT * FROM users WHERE username LIKE %s OR email LIKE %s """ try: param = f"%{keyword}%" cursor.execute(search_sql, (param, param)) return cursor.fetchall() except mysql.connector.Error as err: print(f"搜索失败: {err}") return [] finally: cursor.close()

2.4 更新和删除数据

def update_user_email(conn, user_id, new_email): cursor = conn.cursor() update_sql = "UPDATE users SET email = %s WHERE id = %s" try: cursor.execute(update_sql, (new_email, user_id)) conn.commit() print(f"更新了 {cursor.rowcount} 条记录") except mysql.connector.Error as err: print(f"更新失败: {err}") finally: cursor.close() def delete_user(conn, user_id): cursor = conn.cursor() delete_sql = "DELETE FROM users WHERE id = %s" try: cursor.execute(delete_sql, (user_id,)) conn.commit() print(f"删除了 {cursor.rowcount} 条记录") except mysql.connector.Error as err: print(f"删除失败: {err}") finally: cursor.close()

3. 高级操作与性能优化

3.1 事务处理

MySQL事务是保证数据一致性的重要机制：

def transfer_funds(conn, from_account, to_account, amount): cursor = conn.cursor() try: # 开始事务 conn.start_transaction() # 扣除转出账户金额 cursor.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", (amount, from_account)) # 增加转入账户金额 cursor.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s", (amount, to_account)) # 记录交易 cursor.execute(""" INSERT INTO transactions (from_account, to_account, amount, created_at) VALUES (%s, %s, %s, NOW()) """, (from_account, to_account, amount)) # 提交事务 conn.commit() print("转账成功") except mysql.connector.Error as err: # 回滚事务 conn.rollback() print(f"转账失败，已回滚: {err}") finally: cursor.close()

3.2 连接池管理

对于Web应用等高频访问数据库的场景，使用连接池可以显著提高性能：

from mysql.connector import pooling # 创建连接池 dbconfig = { "host": "localhost", "user": "your_username", "password": "your_password", "database": "your_database" } connection_pool = pooling.MySQLConnectionPool( pool_name="mypool", pool_size=5, # 连接池大小 **dbconfig ) # 使用连接池 def get_user_with_pool(user_id): try: conn = connection_pool.get_connection() cursor = conn.cursor(dictionary=True) cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,)) return cursor.fetchone() except mysql.connector.Error as err: print(f"查询失败: {err}") return None finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # 实际上是返回到连接池

3.3 批量操作优化

对于大量数据操作，使用批量处理可以显著提高性能：

def batch_insert_performance(conn): cursor = conn.cursor() # 方法1：单条插入（慢） start = time.time() for i in range(1000): cursor.execute("INSERT INTO test_table (value) VALUES (%s)", (i,)) conn.commit() print(f"单条插入耗时: {time.time() - start:.2f}秒") # 方法2：批量插入（快） start = time.time() data = [(i,) for i in range(1000, 2000)] cursor.executemany("INSERT INTO test_table (value) VALUES (%s)", data) conn.commit() print(f"批量插入耗时: {time.time() - start:.2f}秒") cursor.close()

4. 常见问题与解决方案

4.1 连接超时问题

MySQL服务器默认会在8小时不活动后断开连接，解决方案：

# 方法1：在连接配置中添加自动重连参数 config = { 'host': 'localhost', 'user': 'your_username', 'password': 'your_password', 'database': 'your_database', 'pool_reset_session': True, # 连接池自动重置会话 'connection_timeout': 30, # 连接超时时间(秒) 'connect_timeout': 10 # 连接建立超时时间(秒) } # 方法2：定期执行简单查询保持连接活跃 def keep_alive(conn): cursor = conn.cursor() try: cursor.execute("SELECT 1") except: # 如果连接已断开，尝试重新连接 conn.reconnect() finally: cursor.close()

4.2 字符编码问题

确保数据库、表和连接都使用utf8mb4编码以支持完整的Unicode字符（包括emoji）：

# 创建表时指定字符集 CREATE TABLE messages ( id INT AUTO_INCREMENT PRIMARY KEY, content TEXT ) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; # 连接配置中添加字符集参数 config = { 'charset': 'utf8mb4', 'collation': 'utf8mb4_unicode_ci' }

4.3 SQL注入防护

始终使用参数化查询，不要直接拼接SQL字符串：

# 错误做法（有SQL注入风险） username = "admin' -- " sql = f"SELECT * FROM users WHERE username = '{username}'" # 正确做法（使用参数化查询） sql = "SELECT * FROM users WHERE username = %s" cursor.execute(sql, (username,))

4.4 性能监控与优化

使用EXPLAIN分析查询性能：

def explain_query(conn, query, params=None): cursor = conn.cursor(dictionary=True) try: cursor.execute(f"EXPLAIN {query}", params or ()) return cursor.fetchall() except mysql.connector.Error as err: print(f"EXPLAIN失败: {err}") return [] finally: cursor.close()

5. 实战项目：简易博客系统

5.1 数据库设计

def setup_blog_database(conn): cursor = conn.cursor() # 创建用户表 cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, password_hash VARCHAR(255) NOT NULL, email VARCHAR(100) NOT NULL UNIQUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_login DATETIME ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) # 创建文章表 cursor.execute(""" CREATE TABLE IF NOT EXISTS posts ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, title VARCHAR(255) NOT NULL, content TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, is_published BOOLEAN DEFAULT FALSE, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) # 创建评论表 cursor.execute(""" CREATE TABLE IF NOT EXISTS comments ( id INT AUTO_INCREMENT PRIMARY KEY, post_id INT NOT NULL, user_id INT NOT NULL, content TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (post_id) REFERENCES posts(id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) conn.commit() cursor.close()

5.2 核心功能实现

class BlogDatabase: def __init__(self, config): self.config = config self.connection_pool = pooling.MySQLConnectionPool( pool_name="blog_pool", pool_size=5, **config ) def get_connection(self): return self.connection_pool.get_connection() def create_user(self, username, email, password_hash): conn = self.get_connection() cursor = conn.cursor() try: cursor.execute(""" INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s) """, (username, email, password_hash)) conn.commit() return cursor.lastrowid except mysql.connector.Error as err: print(f"创建用户失败: {err}") return None finally: cursor.close() conn.close() def create_post(self, user_id, title, content, is_published=False): conn = self.get_connection() cursor = conn.cursor() try: cursor.execute(""" INSERT INTO posts (user_id, title, content, is_published) VALUES (%s, %s, %s, %s) """, (user_id, title, content, is_published)) conn.commit() return cursor.lastrowid except mysql.connector.Error as err: print(f"创建文章失败: {err}") return None finally: cursor.close() conn.close() def get_recent_posts(self, limit=10, published_only=True): conn = self.get_connection() cursor = conn.cursor(dictionary=True) try: query = """ SELECT p.*, u.username FROM posts p JOIN users u ON p.user_id = u.id """ if published_only: query += " WHERE p.is_published = TRUE" query += " ORDER BY p.created_at DESC LIMIT %s" cursor.execute(query, (limit,)) return cursor.fetchall() except mysql.connector.Error as err: print(f"获取文章失败: {err}") return [] finally: cursor.close() conn.close() def add_comment(self, post_id, user_id, content): conn = self.get_connection() cursor = conn.cursor() try: cursor.execute(""" INSERT INTO comments (post_id, user_id, content) VALUES (%s, %s, %s) """, (post_id, user_id, content)) conn.commit() return cursor.lastrowid except mysql.connector.Error as err: print(f"添加评论失败: {err}") return None finally: cursor.close() conn.close()

5.3 性能优化建议

  1. 索引优化：为常用查询条件添加索引
# 在初始化时创建索引 cursor.execute("CREATE INDEX idx_posts_user_id ON posts(user_id)") cursor.execute("CREATE INDEX idx_posts_created_at ON posts(created_at)") cursor.execute("CREATE INDEX idx_comments_post_id ON comments(post_id)")
  1. 分页查询：使用LIMIT和OFFSET实现高效分页
def get_posts_paginated(self, page=1, per_page=10): conn = self.get_connection() cursor = conn.cursor(dictionary=True) try: offset = (page - 1) * per_page cursor.execute(""" SELECT p.*, u.username FROM posts p JOIN users u ON p.user_id = u.id WHERE p.is_published = TRUE ORDER BY p.created_at DESC LIMIT %s OFFSET %s """, (per_page, offset)) return cursor.fetchall() finally: cursor.close() conn.close()
  1. 缓存策略：对热点数据实现缓存机制
from functools import lru_cache @lru_cache(maxsize=100) def get_post_cached(self, post_id): conn = self.get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT p.*, u.username FROM posts p JOIN users u ON p.user_id = u.id WHERE p.id = %s """, (post_id,)) return cursor.fetchone() finally: cursor.close() conn.close()
http://www.jsqmd.com/news/1199845/

相关文章：

  • Windows文件扩展名安全与操作全解析
  • 如何免费在线查看SQLite数据库？这个浏览器工具让你3分钟上手
  • Windows系统光标自定义：从原理到实践的完整指南
  • 互联网大厂 Java 求职面试：从视频直播到微服务的挑战
  • Java集成友盟推送实战与优化指南
  • AI学习工具实战：xiaozhiAI应用与效率提升指南
  • 电商评论数据分析系统开发与情感分析实践
  • 三层交换机VLAN间通信与隔离：从基础互通到精细化ACL策略实战
  • 从零开始玩转业余无线电：新手快速上手指南
  • 音频发生器原理与硬件设计实战指南
  • Sopro TTS模型架构解析：轻量级设计背后的技术原理
  • VS Code本地接入DeepSeek V4 Pro替代Claude Code
  • 实测徐汇浦东长宁多家回收商！百达翡丽 Calatrava 裸表差价上千，哪家无盒证按实时行情估价 - 融媒生活
  • Linux系统运维实战：从环境搭建到性能调优
  • 数据库连接池配置陷阱与性能调优实战
  • 深入理解bpftrace与eBPF事件探测技术
  • Claude Code子代理机制：提升AI开发效率的并行处理方案
  • 终极Blender到虚幻引擎数据转换指南：免费Datasmith插件完整教程
  • Xen虚拟化架构详解：OpenEuler环境下的核心组件与工作原理
  • AI编程助手协作架构：多Agent分工提升开发效率
  • Python操作MySQL数据库实战指南
  • 2026南宁外墙渗水维修怎么处理？对比3家正规公司报价与施工方案（7月） - 防水企业百科
  • 最新AI量化工具推荐，先算清人工确认成本
  • FPGA图像处理核心：滑动窗口模块的通用化设计与资源优化策略
  • AI编程中的Skill与插件：本质区别与18个高效Skill解析
  • DDPG算法解析：连续动作空间强化学习实战指南
  • Google A2UI：AI驱动的动态UI生成技术解析
  • HoRain云--Hermes Agent 技能（Skills）
  • Bril文本格式与JSON格式转换：实用工具详解
  • NTT快速数论变换：从原理到C++实现与优化