Python如何连接MySQL数据库？PyMySQL实战教程

pip install PyMySQL

import pymysql

# 数据库连接参数
db_config = {
    'host': '127.0.0.1', # 或者你的MySQL服务器IP地址
    'port': 3306,
    'user': 'your_username',
    'password': 'your_password',
    'db': 'your_database_name',
    'charset': 'utf8mb4', # 推荐使用utf8mb4，避免字符编码问题
    'cursorclass': pymysql.cursors.DictCursor # 这样查询结果会以字典形式返回，方便处理
}

connection = None
cursor = None

try:
    # 建立数据库连接
    connection = pymysql.connect(**db_config)
    print("数据库连接成功！")

    # 获取游标对象，用于执行SQL查询
    # cursorclass=pymysql.cursors.DictCursor 会让查询结果以字典形式返回
    cursor = connection.cursor()

    # --- 示例1: 创建表 (如果不存在) ---
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(255) NOT NULL,
        email VARCHAR(255) UNIQUE NOT NULL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    """
    cursor.execute(create_table_sql)
    print("表 'users' 检查或创建完成。")

    # --- 示例2: 插入数据 ---
    insert_sql = "INSERT INTO users (name, email) VALUES (%s, %s)"
    user_data = [
        ('张三', 'zhangsan@example.com'),
        ('李四', 'lisi@example.com')
    ]
    cursor.executemany(insert_sql, user_data) # 批量插入
    connection.commit() # 提交事务，保存更改
    print(f"成功插入 {cursor.rowcount} 条数据。")

    # --- 示例3: 查询数据 ---
    select_sql = "SELECT id, name, email FROM users WHERE name = %s"
    cursor.execute(select_sql, ('张三',))
    result = cursor.fetchone() # 获取一条结果
    if result:
        print("\n查询到张三的信息:")
        print(f"ID: {result['id']}, 姓名: {result['name']}, 邮箱: {result['email']}")

    # 查询所有数据
    cursor.execute("SELECT id, name, email FROM users")
    all_users = cursor.fetchall() # 获取所有结果
    print("\n所有用户信息:")
    for user in all_users:
        print(f"ID: {user['id']}, 姓名: {user['name']}, 邮箱: {user['email']}")

    # --- 示例4: 更新数据 ---
    update_sql = "UPDATE users SET email = %s WHERE name = %s"
    cursor.execute(update_sql, ('zhangsan_new@example.com', '张三'))
    connection.commit()
    print(f"\n成功更新 {cursor.rowcount} 条数据。")

    # --- 示例5: 删除数据 ---
    delete_sql = "DELETE FROM users WHERE name = %s"
    cursor.execute(delete_sql, ('李四',))
    connection.commit()
    print(f"成功删除 {cursor.rowcount} 条数据。")

except pymysql.Error as e:
    print(f"数据库操作失败: {e}")
    if connection:
        connection.rollback() # 发生错误时回滚事务
except Exception as e:
    print(f"发生未知错误: {e}")
finally:
    # 无论成功失败，确保关闭游标和连接
    if cursor:
        cursor.close()
    if connection:
        connection.close()
    print("数据库连接已关闭。")

try...except...finally

connection.commit()

utf8

latin1

charset='utf8mb4'

utf8mb4

utf8

host

port

user

password

db

host

127.0.0.1

localhost

INSERT

UPDATE

DELETE

connection.commit()

commit()

sql = "SELECT * FROM users WHERE name = '" + user_input + "'"

cursor.execute(sql, (value1, value2))

cursor.close()

connection.close()

try...finally

with

pymysql.connect

mysql-connector-python

mysql-connector-python

SQLAlchemy

mysql-connector-python

DBUtils

PooledDB

PersistentDB

from DBUtils.PooledDB import PooledDB
import pymysql

# 数据库连接池配置
POOL_CONFIG = {
    'creator': pymysql,  # 指定连接的创建者
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'your_username',
    'password': 'your_password',
    'database': 'your_database_name',
    'charset': 'utf8mb4',
    'maxconnections': 10, # 最大连接数
    'mincached': 2,       # 启动时创建的最小连接数
    'blocking': True,     # 当连接池满时，是否阻塞等待
    'ping': 0,            # 每次获取连接时是否检查连接的有效性，0表示不检查
    'cursorclass': pymysql.cursors.DictCursor
}

# 初始化连接池
pool = PooledDB(**POOL_CONFIG)

def get_db_connection():
    """从连接池获取一个连接"""
    return pool.connection()

# 示例用法
try:
    with get_db_connection() as conn: # with语句会自动关闭连接（返回到池中）
        with conn.cursor() as cursor:
            cursor.execute("SELECT VERSION()")
            version = cursor.fetchone()
            print(f"MySQL 版本: {version['VERSION()']}")

            cursor.execute("SELECT COUNT(*) FROM users")
            count = cursor.fetchone()
            print(f"用户总数: {count['COUNT(*)']}")

except Exception as e:
    print(f"数据库操作失败: {e}")

with

commit()

rollback()

connection.commit()

connection.rollback()