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.connectorconn = 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 MySQLErrortry:# 数据库操作
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()