031-Python 操作 MySQL 数据库

Python 提供了与 MySQL 数据库交互的多种方式，其中最常用的是 mysql-connector 和 PyMySQL 库。通过这些库，可以轻松完成对数据库的连接、查询、插入、更新和删除等操作。

以下是 Python 操作 MySQL 数据库的完整指南。

1. 安装 MySQL 数据库驱动

1.1 安装 mysql-connector

mysql-connector 是 MySQL 官方提供的 Python 驱动。

pip install mysql-connector-python

1.2 安装 PyMySQL

PyMySQL 是一个纯 Python 的 MySQL 驱动，功能强大且易于使用。

pip install PyMySQL

2. 连接 MySQL 数据库

2.1 使用 mysql-connector

  1. import mysql.connector
  2.  
  3. # 连接到 MySQL 数据库
  4. connection = mysql.connector.connect(
  5.     host="localhost",       # MySQL 主机（本地为 localhost）
  6.     user="root",            # MySQL 用户名
  7.     password="password",    # MySQL 密码
  8.     database="test_db"      # 要连接的数据库
  9. )
  10.  
  11. # 检查连接是否成功
  12. if connection.is_connected():
  13.     print("成功连接到数据库")

2.2 使用 PyMySQL

  1. import pymysql
  2.  
  3. # 连接到 MySQL 数据库
  4. connection = pymysql.connect(
  5.     host="localhost",       # MySQL 主机
  6.     user="root",            # 用户名
  7.     password="password",    # 密码
  8.     database="test_db"      # 数据库名
  9. )
  10.  
  11. # 检查连接是否成功
  12. if connection.open:
  13.     print("成功连接到数据库")

3. 执行 SQL 查询

数据库操作通常分为以下几类：

  1. 创建表
  2. 插入数据
  3. 查询数据
  4. 更新数据
  5. 删除数据

以下以 mysql-connector 为例，展示这些操作。如果使用 PyMySQL，代码结构类似。

3.1 创建表

  1. cursor = connection.cursor()
  2.  
  3. # 创建一个名为 users 的表
  4. create_table_query = """
  5. CREATE TABLE IF NOT EXISTS users (
  6.     id INT AUTO_INCREMENT PRIMARY KEY,
  7.     name VARCHAR(50) NOT NULL,
  8.     age INT NOT NULL,
  9.     email VARCHAR(100) UNIQUE NOT NULL
  10. )
  11. """
  12. cursor.execute(create_table_query)
  13. print("表已创建")

3.2 插入数据

单条插入

  1. # 插入单条记录
  2. insert_query = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
  3. data = ("Alice", 25, "[email protected]")
  4. cursor.execute(insert_query, data)
  5.  
  6. # 提交更改
  7. connection.commit()
  8. print("数据已插入")
批量插入

  1. # 插入多条记录
  2. bulk_data = [
  3.     ("Bob", 30, "[email protected]"),
  4.     ("Charlie", 35, "[email protected]"),
  5.     ("David", 40, "[email protected]")
  6. ]
  7. cursor.executemany(insert_query, bulk_data)
  8. connection.commit()
  9. print("批量数据已插入")

3.3 查询数据

查询所有记录

  1. # 查询所有记录
  2. select_query = "SELECT * FROM users"
  3. cursor.execute(select_query)
  4.  
  5. # 获取所有结果
  6. results = cursor.fetchall()
  7. for row in results:
  8.     print(row)
条件查询

  1. # 查询特定条件的数据
  2. condition_query = "SELECT * FROM users WHERE age > %s"
  3. cursor.execute(condition_query, (30,))
  4.  
  5. # 获取结果
  6. results = cursor.fetchall()
  7. for row in results:
  8.     print(row)

3.4 更新数据

  1. # 更新用户的年龄
  2. update_query = "UPDATE users SET age = %s WHERE name = %s"
  3. cursor.execute(update_query, (28, "Alice"))
  4.  
  5. # 提交更改
  6. connection.commit()
  7. print("数据已更新")

3.5 删除数据

  1. # 删除特定用户
  2. delete_query = "DELETE FROM users WHERE name = %s"
  3. cursor.execute(delete_query, ("Bob",))
  4.  
  5. # 提交更改
  6. connection.commit()
  7. print("数据已删除")

4. 关闭连接

完成所有操作后，记得关闭数据库连接。

  1. cursor.close()
  2. connection.close()
  3. print("数据库连接已关闭")

5. 使用上下文管理器

为了保证连接和游标在使用完成后正确关闭，可以使用上下文管理器（with 语句）。

示例：使用上下文管理器

  1. import mysql.connector
  2.  
  3. with mysql.connector.connect(
  4.     host="localhost",
  5.     user="root",
  6.     password="password",
  7.     database="test_db"
  8. ) as connection:
  9.     with connection.cursor() as cursor:
  10.         cursor.execute("SELECT * FROM users")
  11.         for row in cursor.fetchall():
  12.             print(row)

上下文管理器的优势是，即使发生异常，连接和游标也会被正确关闭。

6. 数据库操作完整示例

以下是一个完整的示例，展示了如何创建表、插入数据、查询数据、更新数据和删除数据。

  1. import mysql.connector
  2.  
  3. # 连接到 MySQL 数据库
  4. connection = mysql.connector.connect(
  5.     host="localhost",
  6.     user="root",
  7.     password="password",
  8.     database="test_db"
  9. )
  10.  
  11. cursor = connection.cursor()
  12.  
  13. # 创建表
  14. cursor.execute("""
  15. CREATE TABLE IF NOT EXISTS users (
  16.     id INT AUTO_INCREMENT PRIMARY KEY,
  17.     name VARCHAR(50) NOT NULL,
  18.     age INT NOT NULL,
  19.     email VARCHAR(100) UNIQUE NOT NULL
  20. )
  21. """)
  22.  
  23. # 插入数据
  24. insert_query = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
  25. cursor.execute(insert_query, ("Alice", 25, "[email protected]"))
  26. connection.commit()
  27.  
  28. # 查询数据
  29. cursor.execute("SELECT * FROM users")
  30. for row in cursor.fetchall():
  31.     print(row)
  32.  
  33. # 更新数据
  34. cursor.execute("UPDATE users SET age = %s WHERE name = %s", (26, "Alice"))
  35. connection.commit()
  36.  
  37. # 删除数据
  38. cursor.execute("DELETE FROM users WHERE name = %s", ("Alice",))
  39. connection.commit()
  40.  
  41. # 关闭连接
  42. cursor.close()
  43. connection.close()

7. 处理异常

在实际应用中，需要处理可能发生的异常，比如连接失败或 SQL 查询失败。

示例：异常处理

  1. import mysql.connector
  2. from mysql.connector import Error
  3.  
  4. try:
  5.     # 尝试连接数据库
  6.     connection = mysql.connector.connect(
  7.         host="localhost",
  8.         user="root",
  9.         password="password",
  10.         database="test_db"
  11.     )
  12.  
  13.     if connection.is_connected():
  14.         print("成功连接到数据库")
  15.         cursor = connection.cursor()
  16.  
  17.         # 执行查询
  18.         cursor.execute("SELECT DATABASE();")
  19.         db_name = cursor.fetchone()
  20.         print("当前数据库：", db_name)
  21.  
  22. except Error as e:
  23.     print("连接数据库失败：", e)
  24.  
  25. finally:
  26.     if connection.is_connected():
  27.         cursor.close()
  28.         connection.close()
  29.         print("数据库连接已关闭")

8. 安全性注意事项

  1. 防止 SQL 注入：

    • 始终使用参数化查询，避免直接拼接 SQL 字符串。
    • 示例：不安全的代码： 

      cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")
    • 修复后的代码： 

      cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))

  2. 使用加密连接：

    • 配置 SSL 以保证数据传输的安全性。

  3. 限制权限：

    • 不要使用具有管理员权限的用户连接数据库。
    • 为应用程序创建专用用户，并限制其权限。

9. 总结

通过 Python 操作 MySQL 数据库，可以轻松实现对数据的管理和操作。以下是关键点：

  • 连接数据库：使用 mysql-connector 或 PyMySQL。
  • 执行操作：包括创建表、插入、查询、更新和删除操作。
  • 使用参数化查询：防止 SQL 注入。
  • 关闭连接：确保使用完毕后正确关闭连接。

对于大型项目，推荐结合 ORM（如 SQLAlchemy 或 Django ORM）简化操作，同时保证代码的可维护性和高效性。

以下进一步扩展 Python 操作 MySQL 数据库 的内容，涵盖了高级主题，如事务处理、连接池、ORM（对象关系映射），以及性能优化和实际应用案例。

10. 事务处理

在 MySQL 中，事务（Transaction）是一组作为单个逻辑单元执行的 SQL 语句。事务的主要特性包括：

  • 原子性（Atomicity）：事务要么全部成功，要么全部失败。
  • 一致性（Consistency）：事务执行后，数据库从一个一致状态转移到另一个一致状态。
  • 隔离性（Isolation）：事务之间彼此独立，互不干扰。
  • 持久性（Durability）：事务一旦提交，结果是永久性的。

10.1 启用事务

MySQL 默认在每条 SQL 语句后自动提交。要使用事务，需显式禁用自动提交（autocommit=False）。

事务处理示例

  1. import mysql.connector
  2. from mysql.connector import Error
  3.  
  4. try:
  5.     # 连接到数据库
  6.     connection = mysql.connector.connect(
  7.         host="localhost",
  8.         user="root",
  9.         password="password",
  10.         database="test_db"
  11.     )
  12.     connection.autocommit = False  # 禁用自动提交
  13.  
  14.     cursor = connection.cursor()
  15.  
  16.     # 执行 SQL 操作
  17.     cursor.execute("INSERT INTO users (name, age, email) VALUES (%s, %s, %s)", 
  18.                    ("John", 30, "[email protected]"))
  19.     cursor.execute("UPDATE users SET age = age + 1 WHERE name = %s", 
  20.                    ("John",))
  21.  
  22.     # 提交事务
  23.     connection.commit()
  24.     print("事务已提交")
  25.  
  26. except Error as e:
  27.     # 回滚事务
  28.     connection.rollback()
  29.     print("事务失败，已回滚：", e)
  30.  
  31. finally:
  32.     if connection.is_connected():
  33.         cursor.close()
  34.         connection.close()
  35.         print("数据库连接已关闭")

11. 连接池

连接 MySQL 数据库的开销较大，尤其是在高并发场景中频繁创建和关闭连接会影响性能。连接池可以复用连接，减少开销。

11.1 使用连接池

mysql-connector 提供了内置的连接池支持。

创建连接池

  1. from mysql.connector import pooling
  2.  
  3. # 创建连接池
  4. connection_pool = pooling.MySQLConnectionPool(
  5.     pool_name="mypool",
  6.     pool_size=5,                # 池中最大连接数
  7.     host="localhost",
  8.     user="root",
  9.     password="password",
  10.     database="test_db"
  11. )
  12.  
  13. # 从连接池获取连接
  14. connection = connection_pool.get_connection()
  15.  
  16. if connection.is_connected():
  17.     print("成功从连接池获取连接")
使用连接池的完整示例

  1. from mysql.connector import pooling
  2.  
  3. # 创建连接池
  4. connection_pool = pooling.MySQLConnectionPool(
  5.     pool_name="mypool",
  6.     pool_size=3,
  7.     host="localhost",
  8.     user="root",
  9.     password="password",
  10.     database="test_db"
  11. )
  12.  
  13. # 从连接池中获取连接
  14. try:
  15.     connection = connection_pool.get_connection()
  16.     cursor = connection.cursor()
  17.  
  18.     # 执行查询
  19.     cursor.execute("SELECT * FROM users")
  20.     for row in cursor.fetchall():
  21.         print(row)
  22.  
  23. finally:
  24.     # 释放连接回连接池
  25.     if connection.is_connected():
  26.         cursor.close()
  27.         connection.close()

12. 使用 ORM（对象关系映射）

ORM（Object-Relational Mapping） 是一种将数据库表映射为 Python 对象的技术，可以简化数据库操作。

12.1 SQLAlchemy

SQLAlchemy 是 Python 最流行的 ORM 库之一，支持高级查询和事务管理。

安装 SQLAlchemy

pip install sqlalchemy
SQLAlchemy 的基本使用
  1. 定义模型

  1. from sqlalchemy import create_engine, Column, Integer, String
  2. from sqlalchemy.ext.declarative import declarative_base
  3. from sqlalchemy.orm import sessionmaker
  4.  
  5. # 创建基础类
  6. Base = declarative_base()
  7.  
  8. # 定义用户表模型
  9. class User(Base):
  10.     __tablename__ = "users"
  11.     id = Column(Integer, primary_key=True, autoincrement=True)
  12.     name = Column(String(50), nullable=False)
  13.     age = Column(Integer, nullable=False)
  14.     email = Column(String(100), unique=True, nullable=False)
  1. 连接数据库

  1. # 创建数据库引擎
  2. engine = create_engine("mysql+mysqlconnector://root:password@localhost/test_db")
  3.  
  4. # 创建所有表
  5. Base.metadata.create_all(engine)
  6.  
  7. # 创建会话
  8. Session = sessionmaker(bind=engine)
  9. session = Session()
  1. 插入数据

  1. # 插入数据
  2. new_user = User(name="Alice", age=25, email="[email protected]")
  3. session.add(new_user)
  4. session.commit()
  5. print("数据已插入")
  1. 查询数据

  1. # 查询所有用户
  2. users = session.query(User).all()
  3. for user in users:
  4.     print(user.name, user.age, user.email)
  1. 更新数据

  1. # 更新用户信息
  2. user = session.query(User).filter_by(name="Alice").first()
  3. user.age = 26
  4. session.commit()
  1. 删除数据

  1. # 删除用户
  2. user = session.query(User).filter_by(name="Alice").first()
  3. session.delete(user)
  4. session.commit()

13. 性能优化

在高并发或大数据量场景中，可以通过以下方法优化性能：

13.1 使用索引

为查询频繁的字段创建索引，可以显著提升查询速度。

示例：创建索引

CREATE INDEX idx_name ON users(name);

13.2 分页查询

对于大数据表，避免一次性查询所有数据，使用分页查询。

示例：分页查询

  1. # 分页查询
  2. page = 1
  3. page_size = 10
  4. offset = (page - 1) * page_size
  5.  
  6. query = "SELECT * FROM users LIMIT %s OFFSET %s"
  7. cursor.execute(query, (page_size, offset))
  8.  
  9. for row in cursor.fetchall():
  10.     print(row)

13.3 使用批量插入

对于大量数据插入，使用批量插入代替逐条插入。

  1. # 批量插入
  2. data = [
  3.     ("Tom", 30, "[email protected]"),
  4.     ("Jerry", 29, "[email protected]"),
  5.     ("Spike", 35, "[email protected]")
  6. ]
  7. insert_query = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
  8. cursor.executemany(insert_query, data)
  9. connection.commit()

14. 实际案例：用户管理系统

以下是一个完整的用户管理系统示例，演示如何使用 Python 操作 MySQL 数据库。

功能
  1. 添加用户
  2. 查询所有用户
  3. 更新用户信息
  4. 删除用户

代码实现

  1. import mysql.connector
  2.  
  3. # 数据库连接
  4. connection = mysql.connector.connect(
  5.     host="localhost",
  6.     user="root",
  7.     password="password",
  8.     database="test_db"
  9. )
  10. cursor = connection.cursor()
  11.  
  12. # 创建用户表
  13. cursor.execute("""
  14. CREATE TABLE IF NOT EXISTS users (
  15.     id INT AUTO_INCREMENT PRIMARY KEY,
  16.     name VARCHAR(50),
  17.     age INT,
  18.     email VARCHAR(100) UNIQUE
  19. )
  20. """)
  21.  
  22. def add_user(name, age, email):
  23.     query = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
  24.     cursor.execute(query, (name, age, email))
  25.     connection.commit()
  26.     print(f"用户 {name} 已添加")
  27.  
  28. def get_all_users():
  29.     cursor.execute("SELECT * FROM users")
  30.     for row in cursor.fetchall():
  31.         print(row)
  32.  
  33. def update_user(user_id, name=None, age=None, email=None):
  34.     query = "UPDATE users SET name=%s, age=%s, email=%s WHERE id=%s"
  35.     cursor.execute(query, (name, age, email, user_id))
  36.     connection.commit()
  37.     print(f"用户 {user_id} 已更新")
  38.  
  39. def delete_user(user_id):
  40.     query = "DELETE FROM users WHERE id=%s"
  41.     cursor.execute(query, (user_id,))
  42.     connection.commit()
  43.     print(f"用户 {user_id} 已删除")
  44.  
  45. # 测试功能
  46. add_user("Alice", 25, "[email protected]")
  47. add_user("Bob", 30, "[email protected]")
  48.  
  49. print("所有用户：")
  50. get_all_users()
  51.  
  52. update_user(1, name="Alice Johnson", age=26)
  53. delete_user(2)
  54.  
  55. print("最终用户列表：")
  56. get_all_users()
  57.  
  58. # 关闭连接
  59. cursor.close()
  60. connection.close()

15. 总结

  • 基础操作：熟练掌握如何连接 MySQL、创建表、查询、插入、更新和删除数据。
  • 高级功能：使用事务、连接池和 ORM 简化复杂的数据库操作。
  • 性能优化：通过索引、分页查询和批量操作提升数据库性能。
  • 实际案例：将功能模块化应用到实际项目中。

无论是小型项目还是大型系统，结合 Python 和 MySQL 的强大能力，都可以实现高效、稳定的数据库操作！

注：本文转载自blog.csdn.net的小宝哥Code的文章"https://blog.csdn.net/chenby186119/article/details/145354559"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！