python如何使用mysql数据库的示例代码

《Python如何使用MySQL数据库的示例代码》

在Python开发中，MySQL作为最流行的开源关系型数据库之一，因其高性能、稳定性和易用性被广泛采用。无论是Web应用开发、数据分析还是自动化脚本，掌握Python与MySQL的交互都是开发者必备的技能。本文将通过完整的示例代码，详细讲解如何使用Python连接MySQL数据库，执行增删改查（CRUD）操作，并介绍事务管理、连接池优化等进阶技巧。

一、环境准备

在开始之前，需要确保系统已安装MySQL数据库和Python的MySQL驱动。推荐使用mysql-connector-python或PyMySQL，两者均为纯Python实现，兼容性好。

1. 安装MySQL驱动

pip install mysql-connector-python  # 官方驱动
# 或
pip install pymysql  # 第三方驱动，兼容性更好

2. 创建测试数据库

登录MySQL后执行以下SQL创建测试库和表：

CREATE DATABASE python_mysql_demo;
USE python_mysql_demo;

CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    age INT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

二、基础连接与查询

以下示例使用mysql-connector-python展示基础操作，代码结构清晰，适合初学者理解。

1. 连接数据库

import mysql.connector

def create_connection():
    try:
        conn = mysql.connector.connect(
            host="localhost",
            user="your_username",
            password="your_password",
            database="python_mysql_demo"
        )
        print("数据库连接成功")
        return conn
    except mysql.connector.Error as err:
        print(f"连接失败: {err}")
        return None

# 测试连接
connection = create_connection()

2. 执行查询（SELECT）

def query_users(conn):
    cursor = conn.cursor(dictionary=True)  # 返回字典格式结果
    try:
        cursor.execute("SELECT * FROM users")
        users = cursor.fetchall()
        for user in users:
            print(f"ID: {user['id']}, 姓名: {user['name']}, 邮箱: {user['email']}")
    except mysql.connector.Error as err:
        print(f"查询失败: {err}")
    finally:
        cursor.close()

# 调用查询
if connection:
    query_users(connection)

3. 插入数据（INSERT）

def insert_user(conn, name, email, age):
    cursor = conn.cursor()
    sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
    val = (name, email, age)
    try:
        cursor.execute(sql, val)
        conn.commit()  # 提交事务
        print(f"插入成功，ID: {cursor.lastrowid}")
    except mysql.connector.Error as err:
        conn.rollback()  # 回滚事务
        print(f"插入失败: {err}")
    finally:
        cursor.close()

# 插入测试数据
insert_user(connection, "张三", "zhangsan@example.com", 25)

4. 更新数据（UPDATE）

def update_user_age(conn, user_id, new_age):
    cursor = conn.cursor()
    sql = "UPDATE users SET age = %s WHERE id = %s"
    val = (new_age, user_id)
    try:
        cursor.execute(sql, val)
        conn.commit()
        print(f"更新成功，影响行数: {cursor.rowcount}")
    except mysql.connector.Error as err:
        conn.rollback()
        print(f"更新失败: {err}")
    finally:
        cursor.close()

# 更新ID为1的用户年龄
update_user_age(connection, 1, 26)

5. 删除数据（DELETE）

def delete_user(conn, user_id):
    cursor = conn.cursor()
    sql = "DELETE FROM users WHERE id = %s"
    val = (user_id,)
    try:
        cursor.execute(sql, val)
        conn.commit()
        print(f"删除成功，影响行数: {cursor.rowcount}")
    except mysql.connector.Error as err:
        conn.rollback()
        print(f"删除失败: {err}")
    finally:
        cursor.close()

# 删除ID为2的用户
delete_user(connection, 2)

三、进阶技巧

掌握基础操作后，进一步优化代码性能和安全性至关重要。

1. 使用上下文管理器自动关闭连接

from contextlib import contextmanager

@contextmanager
def get_db_connection():
    conn = mysql.connector.connect(
        host="localhost",
        user="your_username",
        password="your_password",
        database="python_mysql_demo"
    )
    try:
        yield conn
    finally:
        conn.close()

# 使用示例
with get_db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM users")
    print(f"用户总数: {cursor.fetchone()[0]}")

2. 参数化查询防止SQL注入

始终使用参数化查询（如上例中的%s占位符），避免直接拼接SQL字符串。

3. 连接池优化（以PyMySQL为例）

from pymysql import connect
from pymysql.cursors import DictCursor
from dbutils.pooled_db import PooledDB

# 创建连接池
pool = PooledDB(
    creator=connect,
    maxconnections=5,
    mincached=2,
    host="localhost",
    user="your_username",
    password="your_password",
    database="python_mysql_demo",
    cursorclass=DictCursor
)

# 从连接池获取连接
def get_pool_connection():
    return pool.connection()

# 使用示例
conn = get_pool_connection()
try:
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM users LIMIT 5")
        print(cursor.fetchall())
finally:
    conn.close()  # 实际是返回到连接池

4. 批量操作提升性能

def batch_insert_users(conn, user_list):
    cursor = conn.cursor()
    sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
    try:
        cursor.executemany(sql, user_list)  # 批量插入
        conn.commit()
        print(f"批量插入成功，影响行数: {cursor.rowcount}")
    except mysql.connector.Error as err:
        conn.rollback()
        print(f"批量插入失败: {err}")
    finally:
        cursor.close()

# 批量插入数据
users = [
    ("李四", "lisi@example.com", 30),
    ("王五", "wangwu@example.com", 28)
]
batch_insert_users(connection, users)

四、完整项目示例

以下是一个完整的用户管理系统示例，包含连接、查询、增删改查和异常处理。

import mysql.connector
from mysql.connector import Error

class MySQLManager:
    def __init__(self, host, user, password, database):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.connection = None

    def connect(self):
        try:
            self.connection = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            print("MySQL连接成功")
        except Error as e:
            print(f"连接错误: {e}")

    def create_table(self):
        try:
            cursor = self.connection.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS users (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(50) NOT NULL,
                    email VARCHAR(100) UNIQUE NOT NULL,
                    age INT
                )
            """)
            print("表创建成功")
        except Error as e:
            print(f"创建表错误: {e}")
        finally:
            cursor.close()

    def insert_user(self, name, email, age):
        try:
            cursor = self.connection.cursor()
            sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
            cursor.execute(sql, (name, email, age))
            self.connection.commit()
            print(f"用户 {name} 插入成功，ID: {cursor.lastrowid}")
        except Error as e:
            self.connection.rollback()
            print(f"插入用户错误: {e}")
        finally:
            cursor.close()

    def get_all_users(self):
        try:
            cursor = self.connection.cursor(dictionary=True)
            cursor.execute("SELECT * FROM users")
            return cursor.fetchall()
        except Error as e:
            print(f"查询用户错误: {e}")
            return []
        finally:
            cursor.close()

    def update_user(self, user_id, name=None, email=None, age=None):
        try:
            cursor = self.connection.cursor()
            set_clause = []
            params = []
            
            if name:
                set_clause.append("name = %s")
                params.append(name)
            if email:
                set_clause.append("email = %s")
                params.append(email)
            if age is not None:
                set_clause.append("age = %s")
                params.append(age)
            
            if not set_clause:
                print("没有提供更新字段")
                return False
            
            params.append(user_id)
            sql = f"UPDATE users SET {', '.join(set_clause)} WHERE id = %s"
            cursor.execute(sql, params)
            self.connection.commit()
            print(f"更新成功，影响行数: {cursor.rowcount}")
            return True
        except Error as e:
            self.connection.rollback()
            print(f"更新用户错误: {e}")
            return False
        finally:
            cursor.close()

    def delete_user(self, user_id):
        try:
            cursor = self.connection.cursor()
            cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
            self.connection.commit()
            print(f"删除成功，影响行数: {cursor.rowcount}")
        except Error as e:
            self.connection.rollback()
            print(f"删除用户错误: {e}")
        finally:
            cursor.close()

    def close(self):
        if self.connection and self.connection.is_connected():
            self.connection.close()
            print("MySQL连接已关闭")

# 使用示例
if __name__ == "__main__":
    db = MySQLManager(
        host="localhost",
        user="your_username",
        password="your_password",
        database="python_mysql_demo"
    )
    db.connect()
    db.create_table()
    
    # 插入用户
    db.insert_user("赵六", "zhaoliu@example.com", 35)
    
    # 查询所有用户
    users = db.get_all_users()
    print("所有用户:")
    for user in users:
        print(user)
    
    # 更新用户
    db.update_user(1, age=27)
    
    # 删除用户
    db.delete_user(3)
    
    db.close()

五、常见问题与解决方案

1. 连接失败：检查主机名、端口、用户名和密码是否正确，确保MySQL服务已启动。

2. 权限错误：确保用户有对应数据库的访问权限，执行GRANT ALL PRIVILEGES ON python_mysql_demo.* TO 'username'@'localhost';

3. 中文乱码：在连接时指定字符集charset='utf8mb4'。

4. 连接泄漏：始终使用try-finally或上下文管理器确保连接关闭。

关键词：Python、MySQL数据库、连接示例、CRUD操作、事务管理、连接池、参数化查询、SQL注入防护

简介：本文详细介绍了Python如何连接和操作MySQL数据库，包含环境准备、基础CRUD操作、参数化查询防止SQL注入、连接池优化等进阶技巧，并提供完整的用户管理系统示例代码，适合Python开发者快速掌握MySQL数据库交互。