如何使用python对数据库(mysql)进行操作

《如何使用Python对数据库(MySQL)进行操作》

在现代软件开发中，数据库是存储和管理数据的核心组件。MySQL作为开源关系型数据库的代表，因其高性能、易用性和跨平台特性被广泛采用。而Python凭借其简洁的语法和丰富的库生态，成为与MySQL交互的主流编程语言。本文将系统介绍如何使用Python操作MySQL数据库，涵盖环境搭建、基础CRUD操作、高级功能及最佳实践，帮助开发者快速掌握数据库编程技能。

一、环境准备与连接配置

1.1 安装必要组件

使用Python操作MySQL前需安装以下组件：

• MySQL服务器（本地或远程）
• Python MySQL连接器（推荐使用mysql-connector-python或PyMySQL）

通过pip安装连接器：

pip install mysql-connector-python  # 官方驱动
# 或
pip install pymysql                # 第三方驱动

1.2 建立数据库连接

使用mysql.connector建立连接的示例：

import mysql.connector

# 配置连接参数
config = {
    'user': 'your_username',
    'password': 'your_password',
    'host': '127.0.0.1',
    'database': 'your_database',
    'raise_on_warnings': True
}

try:
    # 创建连接对象
    conn = mysql.connector.connect(**config)
    print("MySQL连接成功")
except mysql.connector.Error as err:
    print(f"连接失败: {err}")
finally:
    if 'conn' in locals():
        conn.close()

关键参数说明：

• user/password: 数据库认证信息
• host: 服务器地址（本地可用localhost）
• database: 默认数据库名
• port: 默认3306（可选指定）

二、基础CRUD操作

2.1 创建表结构

使用CREATE TABLE语句创建数据表：

def create_table():
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    
    try:
        cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(50) NOT NULL,
            email VARCHAR(100) UNIQUE,
            age INT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """)
        conn.commit()
        print("表创建成功")
    except mysql.connector.Error as err:
        print(f"创建失败: {err}")
    finally:
        cursor.close()
        conn.close()

2.2 插入数据

单条插入示例：

def insert_user(name, email, age):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    
    sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
    values = (name, email, age)
    
    try:
        cursor.execute(sql, values)
        conn.commit()
        print(f"插入成功，ID: {cursor.lastrowid}")
    except mysql.connector.Error as err:
        print(f"插入失败: {err}")
    finally:
        cursor.close()
        conn.close()

# 批量插入优化
def batch_insert(users_data):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    
    sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
    try:
        cursor.executemany(sql, users_data)
        conn.commit()
        print(f"批量插入{cursor.rowcount}条记录")
    except mysql.connector.Error as err:
        print(f"批量插入失败: {err}")
    finally:
        cursor.close()
        conn.close()

2.3 查询数据

基础查询示例：

def query_users(age_threshold=None):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor(dictionary=True)  # 返回字典格式结果
    
    try:
        if age_threshold:
            sql = "SELECT * FROM users WHERE age > %s"
            cursor.execute(sql, (age_threshold,))
        else:
            cursor.execute("SELECT * FROM users")
        
        results = cursor.fetchall()
        for user in results:
            print(f"ID: {user['id']}, Name: {user['name']}, Age: {user['age']}")
        return results
    except mysql.connector.Error as err:
        print(f"查询失败: {err}")
    finally:
        cursor.close()
        conn.close()

2.4 更新与删除

def update_user(user_id, new_email):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    
    sql = "UPDATE users SET email = %s WHERE id = %s"
    try:
        cursor.execute(sql, (new_email, user_id))
        conn.commit()
        print(f"更新{cursor.rowcount}条记录")
    except mysql.connector.Error as err:
        print(f"更新失败: {err}")
    finally:
        cursor.close()
        conn.close()

def delete_user(user_id):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    
    sql = "DELETE FROM users WHERE id = %s"
    try:
        cursor.execute(sql, (user_id,))
        conn.commit()
        print(f"删除{cursor.rowcount}条记录")
    except mysql.connector.Error as err:
        print(f"删除失败: {err}")
    finally:
        cursor.close()
        conn.close()

三、高级功能实现

3.1 事务处理

def transfer_funds(from_id, to_id, amount):
    conn = mysql.connector.connect(**config)
    try:
        # 开启事务
        conn.start_transaction()
        
        cursor1 = conn.cursor()
        cursor2 = conn.cursor()
        
        # 扣减转出账户
        cursor1.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_id)
        )
        
        # 增加转入账户
        cursor2.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_id)
        )
        
        conn.commit()
        print("转账成功")
    except mysql.connector.Error as err:
        conn.rollback()
        print(f"转账失败: {err}")
    finally:
        if 'conn' in locals():
            conn.close()

3.2 存储过程调用

def call_stored_procedure():
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    
    try:
        # 调用存储过程
        args = (10, 0)  # 参数：年龄阈值，输出计数
        cursor.callproc('count_users_by_age', args)
        
        # 获取结果（存储过程第二个参数为输出）
        print(f"符合条件的用户数: {args[1]}")
    except mysql.connector.Error as err:
        print(f"调用失败: {err}")
    finally:
        cursor.close()
        conn.close()

3.3 连接池优化

使用mysql.connector.pooling创建连接池：

from mysql.connector import pooling

dbconfig = {
    "host": "localhost",
    "user": "root",
    "password": "password",
    "database": "testdb"
}

connection_pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=5,
    **dbconfig
)

def get_connection():
    return connection_pool.get_connection()

# 使用示例
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
# ...操作

四、最佳实践与安全建议

4.1 参数化查询防注入

永远不要使用字符串拼接构建SQL语句：

# 危险示例（SQL注入风险）
# email = "user@example.com' OR '1'='1"
# cursor.execute(f"SELECT * FROM users WHERE email = '{email}'")

# 安全做法
cursor.execute("SELECT * FROM users WHERE email = %s", (email,))

4.2 资源管理

• 使用try-finally或上下文管理器确保连接关闭
• 批量操作时合理设置fetchsize控制内存

4.3 性能优化

• 为常用查询字段添加索引
• 避免在循环中频繁创建/关闭连接
• 对大表查询使用分页（LIMIT offset, size）

4.4 日志与错误处理

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

try:
    # 数据库操作
except mysql.connector.Error as err:
    logging.error(f"数据库错误: {err}")
    raise  # 或进行降级处理

五、完整示例：用户管理系统

import mysql.connector
from mysql.connector import Error

class UserManager:
    def __init__(self, config):
        self.config = config
    
    def connect(self):
        try:
            return mysql.connector.connect(**self.config)
        except Error as e:
            print(f"连接错误: {e}")
            return None
    
    def create_user(self, name, email, age):
        conn = self.connect()
        if not conn:
            return False
        
        cursor = conn.cursor()
        sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
        try:
            cursor.execute(sql, (name, email, age))
            conn.commit()
            return True
        except Error as e:
            print(f"创建失败: {e}")
            conn.rollback()
            return False
        finally:
            cursor.close()
            conn.close()
    
    def get_users(self, age_min=None):
        conn = self.connect()
        if not conn:
            return []
        
        cursor = conn.cursor(dictionary=True)
        try:
            if age_min:
                cursor.execute("SELECT * FROM users WHERE age >= %s", (age_min,))
            else:
                cursor.execute("SELECT * FROM users")
            return cursor.fetchall()
        except Error as e:
            print(f"查询错误: {e}")
            return []
        finally:
            cursor.close()
            conn.close()

# 使用示例
if __name__ == "__main__":
    db_config = {
        'host': 'localhost',
        'user': 'app_user',
        'password': 'secure_pass',
        'database': 'app_db'
    }
    
    manager = UserManager(db_config)
    
    # 添加用户
    manager.create_user("张三", "zhangsan@example.com", 28)
    manager.create_user("李四", "lisi@example.com", 32)
    
    # 查询用户
    users = manager.get_users(age_min=30)
    for user in users:
        print(f"{user['name']} ({user['email']}), 年龄: {user['age']}")

关键词

Python、MySQL、数据库操作、CRUD、连接池、事务处理、参数化查询、SQL注入防护、存储过程、连接配置

简介

本文详细介绍了使用Python操作MySQL数据库的完整流程，涵盖环境搭建、基础CRUD操作、事务处理、存储过程调用等高级功能，并提供连接池优化、安全防护等最佳实践。通过代码示例和完整项目案例，帮助开发者快速掌握Python数据库编程技能。