Python MySQL(mysql-connector驱动)
MySQL是广泛使用的关系型数据库，Python通过mysql-connector驱动可以方便地操作MySQL数据库。本文将详细介绍如何使用Python连接和操作MySQL。
安装mysql-connector
首先需要安装MySQL官方提供的Python驱动：
pip install mysql-connector-python验证安装是否成功：
import mysql.connector
print("mysql-connector安装成功")如果没有报错，说明安装成功。
连接MySQL数据库
基本连接
import mysql.connector
# 创建数据库连接
mydb = mysql.connector.connect(
host="localhost", # 数据库地址
user="root", # 用户名
password="123456", # 密码
database="test_db" # 数据库名
)
print("数据库连接成功")处理连接异常
在实际应用中，应该处理连接可能出现的异常：
import mysql.connector
from mysql.connector import Error
try:
mydb = mysql.connector.connect(
host="localhost",
user="root",
password="123456",
database="test_db"
)
if mydb.is_connected():
print("成功连接到MySQL数据库")
except Error as e:
print(f"连接错误: {e}")
finally:
if mydb.is_connected():
mydb.close()
print("数据库连接已关闭")数据库操作
创建数据库
import mysql.connector
mydb = mysql.connector.connect(
host="localhost",
user="root",
password="123456"
)
mycursor = mydb.cursor()
# 创建数据库
mycursor.execute("CREATE DATABASE IF NOT EXISTS school_db")
print("数据库创建成功")查看所有数据库
mycursor.execute("SHOW DATABASES")
print("所有数据库:")
for db in mycursor:
print(f"- {db[0]}")数据表操作
创建数据表
import mysql.connector
mydb = mysql.connector.connect(
host="localhost",
user="root",
password="123456",
database="school_db"
)
mycursor = mydb.cursor()
# 创建学生表
mycursor.execute("""
CREATE TABLE IF NOT EXISTS students (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
age INT,
grade VARCHAR(10),
email VARCHAR(100)
)
""")
print("学生表创建成功")查看数据表
mycursor.execute("SHOW TABLES")
print("所有数据表:")
for table in mycursor:
print(f"- {table[0]}")数据操作
插入单条数据
# 插入一条学生记录
sql = "INSERT INTO students (name, age, grade, email) VALUES (%s, %s, %s, %s)"
val = ("张三", 18, "高三", "zhangsan@fly63.com")
mycursor.execute(sql, val)
mydb.commit() # 提交事务
print(f"插入成功，ID: {mycursor.lastrowid}")批量插入数据
# 批量插入学生记录
sql = "INSERT INTO students (name, age, grade, email) VALUES (%s, %s, %s, %s)"
students = [
("李四", 17, "高二", "lisi@fly63.com"),
("王五", 19, "高三", "wangwu@fly63.com"),
("赵六", 16, "高一", "zhaoliu@fly63.com"),
("钱七", 18, "高三", "qianqi@fly63.com")
]
mycursor.executemany(sql, students)
mydb.commit()
print(f"批量插入成功，共{mycursor.rowcount}条记录")查询数据
查询所有数据
# 查询所有学生
mycursor.execute("SELECT * FROM students")
results = mycursor.fetchall()
print("所有学生信息:")
for row in results:
print(f"ID: {row[0]}, 姓名: {row[1]}, 年龄: {row[2]}, 年级: {row[3]}, 邮箱: {row[4]}")查询指定字段
# 只查询姓名和年龄
mycursor.execute("SELECT name, age FROM students")
results = mycursor.fetchall()
print("学生姓名和年龄:")
for name, age in results:
print(f"{name} - {age}岁")条件查询
# 查询高三学生
sql = "SELECT * FROM students WHERE grade = %s"
val = ("高三",)
mycursor.execute(sql, val)
results = mycursor.fetchall()
print("高三学生:")
for row in results:
print(f"姓名: {row[1]}, 年龄: {row[2]}")使用LIKE模糊查询
# 查询姓名包含"张"的学生
sql = "SELECT * FROM students WHERE name LIKE %s"
val = ("%张%",)
mycursor.execute(sql, val)
results = mycursor.fetchall()
print("姓张的学生:")
for row in results:
print(f"姓名: {row[1]}")数据排序和限制
排序查询
# 按年龄升序排列
mycursor.execute("SELECT * FROM students ORDER BY age ASC")
results = mycursor.fetchall()
print("按年龄升序:")
for row in results:
print(f"姓名: {row[1]}, 年龄: {row[2]}")限制查询结果数量
# 查询前3条记录
mycursor.execute("SELECT * FROM students LIMIT 3")
results = mycursor.fetchall()
print("前3名学生:")
for row in results:
print(f"姓名: {row[1]}")分页查询
# 分页查询，每页2条记录
page = 1
page_size = 2
offset = (page - 1) * page_size
sql = "SELECT * FROM students LIMIT %s OFFSET %s"
val = (page_size, offset)
mycursor.execute(sql, val)
results = mycursor.fetchall()
print(f"第{page}页数据:")
for row in results:
print(f"姓名: {row[1]}")更新数据
更新单条记录
# 更新学生年龄
sql = "UPDATE students SET age = %s WHERE name = %s"
val = (19, "张三")
mycursor.execute(sql, val)
mydb.commit()
print(f"更新成功，影响{mycursor.rowcount}条记录")批量更新
# 批量更新年级
updates = [
(20, "张三"),
(18, "李四")
]
sql = "UPDATE students SET age = %s WHERE name = %s"
mycursor.executemany(sql, updates)
mydb.commit()
print(f"批量更新成功，影响{mycursor.rowcount}条记录")删除数据
删除单条记录
# 删除指定学生
sql = "DELETE FROM students WHERE name = %s"
val = ("赵六",)
mycursor.execute(sql, val)
mydb.commit()
print(f"删除成功，影响{mycursor.rowcount}条记录")清空表数据
# 清空学生表（谨慎使用！）
mycursor.execute("DELETE FROM students")
mydb.commit()
print("所有学生记录已清空")高级操作
使用事务
try:
# 开始事务
mydb.start_transaction()
# 插入新学生
sql1 = "INSERT INTO students (name, age, grade) VALUES (%s, %s, %s)"
val1 = ("孙八", 17, "高二")
mycursor.execute(sql1, val1)
# 更新另一个学生
sql2 = "UPDATE students SET age = %s WHERE name = %s"
val2 = (20, "张三")
mycursor.execute(sql2, val2)
# 提交事务
mydb.commit()
print("事务执行成功")
except Exception as e:
# 回滚事务
mydb.rollback()
print(f"事务执行失败，已回滚: {e}")创建带外键的表
# 创建课程表
mycursor.execute("""
CREATE TABLE IF NOT EXISTS courses (
course_id INT AUTO_INCREMENT PRIMARY KEY,
course_name VARCHAR(100) NOT NULL,
teacher VARCHAR(100)
)
""")
# 创建选课表（带外键）
mycursor.execute("""
CREATE TABLE IF NOT EXISTS student_courses (
id INT AUTO_INCREMENT PRIMARY KEY,
student_id INT,
course_id INT,
score DECIMAL(4,1),
FOREIGN KEY (student_id) REFERENCES students(id),
FOREIGN KEY (course_id) REFERENCES courses(course_id)
)
""")
print("带外键的表创建成功")实际应用示例
学生管理系统
class StudentManager:
def __init__(self):
self.connection = mysql.connector.connect(
host="localhost",
user="root",
password="123456",
database="school_db"
)
self.cursor = self.connection.cursor()
def add_student(self, name, age, grade, email):
"""添加学生"""
sql = "INSERT INTO students (name, age, grade, email) VALUES (%s, %s, %s, %s)"
val = (name, age, grade, email)
self.cursor.execute(sql, val)
self.connection.commit()
return self.cursor.lastrowid
def get_students_by_grade(self, grade):
"""按年级查询学生"""
sql = "SELECT * FROM students WHERE grade = %s"
self.cursor.execute(sql, (grade,))
return self.cursor.fetchall()
def update_student_email(self, student_id, new_email):
"""更新学生邮箱"""
sql = "UPDATE students SET email = %s WHERE id = %s"
self.cursor.execute(sql, (new_email, student_id))
self.connection.commit()
return self.cursor.rowcount
def delete_student(self, student_id):
"""删除学生"""
sql = "DELETE FROM students WHERE id = %s"
self.cursor.execute(sql, (student_id,))
self.connection.commit()
return self.cursor.rowcount
def close(self):
"""关闭连接"""
self.cursor.close()
self.connection.close()
# 使用示例
manager = StudentManager()
# 添加学生
new_id = manager.add_student("周九", 16, "高一", "zhoujiu@fly63.com")
print(f"添加学生成功，ID: {new_id}")
# 查询高二学生
students = manager.get_students_by_grade("高二")
print("高二学生:")
for student in students:
print(f"姓名: {student[1]}")
manager.close()数据库备份工具
import mysql.connector
import datetime
def backup_database(host, user, password, database):
"""备份数据库到文件"""
try:
# 连接数据库
conn = mysql.connector.connect(
host=host,
user=user,
password=password,
database=database
)
cursor = conn.cursor()
# 获取所有表
cursor.execute("SHOW TABLES")
tables = [table[0] for table in cursor.fetchall()]
# 生成备份文件名
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
backup_file = f"backup_{database}_{timestamp}.sql"
with open(backup_file, 'w', encoding='utf-8') as f:
f.write(f"-- 数据库备份: {database}\n")
f.write(f"-- 备份时间: {timestamp}\n\n")
for table in tables:
# 备份表结构
cursor.execute(f"SHOW CREATE TABLE {table}")
create_table_sql = cursor.fetchone()[1]
f.write(f"{create_table_sql};\n\n")
# 备份表数据
cursor.execute(f"SELECT * FROM {table}")
rows = cursor.fetchall()
if rows:
f.write(f"-- 数据表: {table}\n")
for row in rows:
# 简化处理，实际应该更复杂
values = ", ".join([f"'{str(col)}'" for col in row])
f.write(f"INSERT INTO {table} VALUES ({values});\n")
f.write("\n")
print(f"备份成功: {backup_file}")
except Exception as e:
print(f"备份失败: {e}")
finally:
if conn.is_connected():
cursor.close()
conn.close()
# 使用备份功能
backup_database("localhost", "root", "123456", "school_db")性能优化建议
使用连接池：对于Web应用，使用连接池管理数据库连接
批量操作：尽量使用executemany进行批量插入和更新
索引优化：为经常查询的字段创建索引
适时关闭连接：用完数据库连接后及时关闭
使用预处理语句：防止SQL注入，提高性能
错误处理最佳实践
import mysql.connector
from mysql.connector import Error
def safe_database_operation():
try:
conn = mysql.connector.connect(
host="localhost",
user="root",
password="123456",
database="school_db"
)
if conn.is_connected():
cursor = conn.cursor()
# 执行数据库操作
cursor.execute("SELECT * FROM students")
results = cursor.fetchall()
return results
except Error as e:
print(f"数据库错误: {e}")
return None
finally:
# 确保连接被关闭
if 'conn' in locals() and conn.is_connected():
cursor.close()
conn.close()掌握Python操作MySQL数据库是Web开发和数据分析的重要技能。通过本文的介绍，你应该能够完成基本的数据库操作，为实际项目开发打下基础。
本文内容仅供个人学习/研究/参考使用，不构成任何决策建议或专业指导。分享/转载时请标明原文来源，同时请勿将内容用于商业售卖、虚假宣传等非学习用途哦～感谢您的理解与支持！