python操作MySQL数据库

1. 安装依赖库

常用库：

• PyMySQL（纯 Python 实现）

• mysql-connector-python（MySQL 官方驱动）

# 安装 PyMySQL
pip install pymysql

# 安装 mysql-connector-python
pip install mysql-connector-python

2. 连接数据库

使用 PyMySQL

import pymysql

# 建立连接
conn = pymysql.connect(
    host='localhost',   # 数据库地址
    user='root',        # 用户名
    password='123456', # 密码
    database='testdb',  # 数据库名
    charset='utf8mb4',  # 字符编码
    cursorclass=pymysql.cursors.DictCursor  # 返回字典格式的结果
)

# 创建游标对象
cursor = conn.cursor()

使用 mysql-connector

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='123456',
    database='testdb'
)
cursor = conn.cursor(dictionary=True)  # 返回字典格式结果

3. 执行 SQL 操作

创建表

create_table_sql = """
CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255),
    email VARCHAR(255)
)
"""
cursor.execute(create_table_sql)
conn.commit()  # 提交事务

插入数据

# 单条插入（参数化查询防止 SQL 注入）
insert_sql = "INSERT INTO users (name, email) VALUES (%s, %s)"
cursor.execute(insert_sql, ('Alice', 'alice@example.com'))
conn.commit()

# 批量插入
data = [('Bob', 'bob@example.com'), ('Charlie', 'charlie@example.com')]
cursor.executemany(insert_sql, data)
conn.commit()

查询数据

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

# 获取所有结果
results = cursor.fetchall()
for row in results:
    print(row)  # 字典格式（如使用 DictCursor）或元组

# 获取单条结果
single_row = cursor.fetchone()
print(single_row)

更新数据

update_sql = "UPDATE users SET email = %s WHERE name = %s"
cursor.execute(update_sql, ('new_email@example.com', 'Alice'))
conn.commit()

删除数据

delete_sql = "DELETE FROM users WHERE name = %s"
cursor.execute(delete_sql, ('Bob',))
conn.commit()

4. 事务管理

try:
    # 执行多个操作
    cursor.execute(sql1)
    cursor.execute(sql2)
    conn.commit()  # 提交事务
except Exception as e:
    conn.rollback()  # 回滚事务
    print(f"Error: {e}")

5. 关闭连接

cursor.close()
conn.close()

6. 异常处理

from pymysql import MySQLError

try:
    # 数据库操作
except MySQLError as e:
    print(f"MySQL Error: {e}")
except Exception as e:
    print(f"General Error: {e}")

完整示例

import pymysql

# 连接数据库
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='123456',
    database='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

try:
    with conn.cursor() as cursor:
        # 创建表
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(255),
                email VARCHAR(255)
            )
        """)
        
        # 插入数据
        cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ('Alice', 'alice@example.com'))
        
        # 查询数据
        cursor.execute("SELECT * FROM users")
        rows = cursor.fetchall()
        print("Users:")
        for row in rows:
            print(row)
    
    conn.commit()  # 提交事务

except Exception as e:
    conn.rollback()
    print(f"Error: {e}")

finally:
    conn.close()