Python操作MySQL：从基础连接到CRUD实战 1. Python操作MySQL基础环境搭建1.1 MySQL驱动选择与安装Python连接MySQL数据库需要依赖特定的数据库驱动。目前主流的选择有mysql-connector-pythonMySQL官方驱动PyMySQL纯Python实现MySQLdbC扩展实现对于新手而言我推荐使用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(dictionaryTrue) # 返回字典形式的结果 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(dictionaryTrue) 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_namemypool, pool_size5, # 连接池大小 **dbconfig ) # 使用连接池 def get_user_with_pool(user_id): try: conn connection_pool.get_connection() cursor conn.cursor(dictionaryTrue) 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 fSELECT * FROM users WHERE username {username} # 正确做法使用参数化查询 sql SELECT * FROM users WHERE username %s cursor.execute(sql, (username,))4.4 性能监控与优化使用EXPLAIN分析查询性能def explain_query(conn, query, paramsNone): cursor conn.cursor(dictionaryTrue) try: cursor.execute(fEXPLAIN {query}, params or ()) return cursor.fetchall() except mysql.connector.Error as err: print(fEXPLAIN失败: {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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 ) # 创建文章表 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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 ) # 创建评论表 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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 ) conn.commit() cursor.close()5.2 核心功能实现class BlogDatabase: def __init__(self, config): self.config config self.connection_pool pooling.MySQLConnectionPool( pool_nameblog_pool, pool_size5, **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_publishedFalse): 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, limit10, published_onlyTrue): conn self.get_connection() cursor conn.cursor(dictionaryTrue) 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 性能优化建议索引优化为常用查询条件添加索引# 在初始化时创建索引 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))分页查询使用LIMIT和OFFSET实现高效分页def get_posts_paginated(self, page1, per_page10): conn self.get_connection() cursor conn.cursor(dictionaryTrue) 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()缓存策略对热点数据实现缓存机制from functools import lru_cache lru_cache(maxsize100) def get_post_cached(self, post_id): conn self.get_connection() cursor conn.cursor(dictionaryTrue) 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()