数据存储：一文掌握存储数据到mysql的详细使用

文章目录

- 一、环境准备
- - 1.1 安装MySQL数据库
  - 1.2 安装Python MySQL驱动
- 二、连接到MySQL数据库
- 三、执行基本的CRUD操作
- - 3.1 创建（Create）：插入数据
  - 3.2 读取（Read）：查询数据
  - 3.3 更新（Update）：更新数据
  - 3.4 删除（Delete）：删除数据
  - 3.5 完整代码示例
- 四、使用上下文管理器简化连接和游标管理
- 五、处理事务
- 六、使用ORM框架（推荐）
- 七、最佳实践
- 八、总结

在数据采集、处理和存储过程中，MySQL作为一种广泛使用的关系型数据库管理系统，因其高性能、可靠性和易用性而备受青睐。Python提供了多种库来连接和操作MySQL数据库，其中最常用的是sql>mysql-connector-python和PyMySQL。本文将详细介绍如何使用Python连接MySQL数据库，执行基本的CRUD（创建、读取、更新、删除）操作，并提供相应的代码示例。

一、环境准备

1.1 安装MySQL数据库

确保已经在系统中安装并运行了MySQL数据库。可以从 MySQL官方网站下载并安装适合你操作系统的版本。

1.2 安装Python MySQL驱动

使用pip安装sql>mysql-connector-python或PyMySQL库。这里以PyMySQL为例：

pip install PyMySQL

二、连接到MySQL数据库

首先，需要使用Python代码连接到MySQL数据库。以下是使用PyMySQL连接数据库的基本步骤：

python">import pysql>mysql

# 连接参数
connection = pysql>mysql.connect(
    host='localhost',        # 数据库主机地址
    user='your_username',    # 数据库用户名
    password='your_password',# 数据库密码
    database='your_database',# 要连接的数据库名称
    charset='utf8mb4',       # 字符集
    cursorclass=pysql>mysql.cursors.DictCursor  # 返回字典类型的结果
)

try:
    with connection.cursor() as cursor:
        # 在此处执行SQL语句
        pass
finally:
    connection.close()

三、执行基本的CRUD操作

3.1 创建（Create）：插入数据

使用INSERT INTO语句向表中插入数据。

python">import pysql>mysql

connection = pysql>mysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='test_db',
    charset='utf8mb4',
    cursorclass=pysql>mysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
        cursor.execute(sql, ('张三', 28, 'zhangsan@example.com'))
    connection.commit()  # 提交事务
finally:
    connection.close()

3.2 读取（Read）：查询数据

使用SELECT语句从表中查询数据。

python">import pysql>mysql

connection = pysql>mysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='test_db',
    charset='utf8mb4',
    cursorclass=pysql>mysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        sql = "SELECT id, name, age, email FROM users WHERE age > %s"
        cursor.execute(sql, (25,))
        results = cursor.fetchall()
        for row in results:
            print(row)
finally:
    connection.close()

3.3 更新（Update）：更新数据

使用UPDATE语句修改表中的数据。

python">import pysql>mysql

connection = pysql>mysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='test_db',
    charset='utf8mb4',
    cursorclass=pysql>mysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        sql = "UPDATE users SET age = %s WHERE name = %s"
        cursor.execute(sql, (29, '张三'))
    connection.commit()
finally:
    connection.close()

3.4 删除（Delete）：删除数据

使用DELETE语句删除表中的数据。

python">import pysql>mysql

connection = pysql>mysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='test_db',
    charset='utf8mb4',
    cursorclass=pysql>mysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        sql = "DELETE FROM users WHERE name = %s"
        cursor.execute(sql, ('李四',))
    connection.commit()
finally:
    connection.close()

3.5 完整代码示例

以下是一个完整的Python脚本，演示了如何连接MySQL数据库并执行CRUD操作。

python">import pysql>mysql

# 数据库连接配置
config = {
    'host': 'localhost',        # 数据库主机地址
    'port': 3306,               # 数据库端口，默认3306
    'user': 'your_username',    # 数据库用户名
    'password': 'your_password',# 数据库密码
    'database': 'test_db',      # 要连接的数据库名称
    'charset': 'utf8mb4',       # 字符集
    'cursorclass': pysql>mysql.cursors.DictCursor  # 返回字典类型的结果
}

def create_database(cursor):
    """创建数据库"""
    cursor.execute("CREATE DATABASE IF NOT EXISTS test_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;")
    print("数据库创建成功或已存在。")

def create_table(cursor):
    """创建表"""
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(100) NOT NULL,
        age INT,
        email VARCHAR(150) UNIQUE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    """
    cursor.execute(create_table_sql)
    print("表创建成功或已存在。")

def insert_data(cursor, connection, data):
    """插入数据"""
    sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
    cursor.execute(sql, data)
    connection.commit()
    print(f"插入一条数据，ID: {cursor.lastrowid}")

def query_data(cursor):
    """查询数据"""
    sql = "SELECT id, name, age, email FROM users WHERE age > %s"
    cursor.execute(sql, (25,))
    results = cursor.fetchall()
    print("查询结果:")
    for row in results:
        print(row)

def update_data(cursor, connection, user_id, new_age):
    """更新数据"""
    sql = "UPDATE users SET age = %s WHERE id = %s"
    cursor.execute(sql, (new_age, user_id))
    connection.commit()
    print(f"更新ID为{user_id}的用户年龄为{new_age}")

def delete_data(cursor, connection, user_id):
    """删除数据"""
    sql = "DELETE FROM users WHERE id = %s"
    cursor.execute(sql, (user_id,))
    connection.commit()
    print(f"删除ID为{user_id}的用户")

def main():
    try:
        # 连接到MySQL服务器
        connection = pysql>mysql.connect(**config)
        with connection.cursor() as cursor:
            # 创建数据库
            create_database(cursor)
            
            # 使用刚创建的数据库
            connection.select_db('test_db')
            
            # 创建表
            create_table(cursor)
            
            # 插入数据
            users = [
                ('张三', 28, 'zhangsan@example.com'),
                ('李四', 34, 'lisi@example.com'),
                ('王五', 23, 'wangwu@example.com')
            ]
            for user in users:
                insert_data(cursor, connection, user)
            
            # 查询数据
            print("\n所有用户信息:")
            query_data(cursor)
            
            # 更新数据
            update_data(cursor, connection, 1, 29)
            
            # 查询更新后的数据
            print("\n更新后的用户信息:")
            query_data(cursor)
            
            # 删除数据
            delete_data(cursor, connection, 2)
            
            # 查询删除后的数据
            print("\n删除后的用户信息:")
            query_data(cursor)
    
    except pysql>mysql.MySQLError as e:
        print(f"发生错误: {e}")
    finally:
        if 'connection' in locals() and connection.open:
            connection.close()
            print("\n数据库连接已关闭。")

if __name__ == "__main__":
    main()

四、使用上下文管理器简化连接和游标管理

为了简化代码和提高可读性，可以使用上下文管理器（with语句）来自动管理数据库连接和游标。

python">import pysql>mysql

connection_params = {
    'host': 'localhost',
    'user': 'your_username',
    'password': 'your_password',
    'database': 'test_db',
    'charset': 'utf8mb4',
    'cursorclass': pysql>mysql.cursors.DictCursor
}

sql_insert = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
data = ('王五', 25, 'wangwu@example.com')

with pysql>mysql.connect(**connection_params) as connection:
    with connection.cursor() as cursor:
        cursor.execute(sql_insert, data)
    connection.commit()

五、处理事务

事务是确保数据库操作原子性、一致性、隔离性和持久性（ACID）的重要机制。PyMySQL支持事务管理，可以通过commit()和rollback()方法来控制事务。

python">import pysql>mysql

connection_params = {
    'host': 'localhost',
    'user': 'your_username',
    'password': 'your_password',
    'database': 'test_db',
    'charset': 'utf8mb4',
    'cursorclass': pysql>mysql.cursors.DictCursor
}

sql_insert = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
sql_update = "UPDATE users SET age = %s WHERE name = %s"

data_insert = ('赵六', 30, 'zhaoliu@example.com')
data_update = (31, '赵六')

try:
    with pysql>mysql.connect(**connection_params) as connection:
        with connection.cursor() as cursor:
            cursor.execute(sql_insert, data_insert)
            cursor.execute(sql_update, data_update)
        connection.commit()  # 提交事务
except Exception as e:
    connection.rollback()  # 回滚事务
    print(f"发生错误: {e}")

六、使用ORM框架（推荐）

虽然直接使用SQL语句可以灵活地操作数据库，但对于复杂的项目，使用ORM（对象关系映射）框架可以提高开发效率和代码可维护性。SQLAlchemy是Python中一个强大的ORM框架，支持多种数据库，包括MySQL。

安装SQLAlchemy

pip install SQLAlchemy pysql>mysql

使用SQLAlchemy连接MySQL

python">from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# 创建数据库引擎
engine = create_engine('sql>mysql+pysql>mysql://your_username:your_password@localhost/test_db?charset=utf8mb4')

# 创建基类
Base = declarative_base()

# 定义User模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))
    age = Column(Integer)
    email = Column(String(100))

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 创建表（如果不存在）
Base.metadata.create_all(engine)

# 添加新用户
new_user = User(name='孙七', age=27, email='sunqi@example.com')
session.add(new_user)
session.commit()

# 查询用户
users = session.query(User).filter(User.age > 25).all()
for user in users:
    print(user.name, user.age, user.email)

# 关闭会话
session.close()

七、最佳实践

1、使用参数化查询：避免SQL注入攻击，使用参数化查询（如 %s 占位符）或ORM框架。

2、管理连接池：对于高并发应用，使用连接池（如SQLAlchemy的连接池）来管理数据库连接。

3、处理异常：建议使用 try-except 捕获并处理数据库操作中的异常，确保程序的健壮性。

4、关闭连接：确保在操作完成后关闭数据库连接，释放资源。

5、优化查询：编写高效的SQL查询，避免全表扫描和不必要的复杂查询。

6、事务管理：在执行插入、更新、删除操作后，需要调用 conn.commit() 提交事务。

八、总结

使用Python连接和操作MySQL数据库是数据采集、处理和存储中的常见需求。通过PyMySQL、sql>mysql-connector-python等库，可以方便地执行CRUD操作；而使用ORM框架如SQLAlchemy，则可以提高开发效率和代码的可维护性。在实际应用中，结合具体需求选择合适的方法，并遵循最佳实践，可以确保数据库操作的效率和安全性。