Python 接入 MySQL 数据库

安装驱动

pip install pymysql cryptography

基本操作流程

1. connect() → 创建连接
2. cursor() → 获取游标
3. execute() → 执行 SQL
4. commit() / rollback() → 提交/回滚事务
5. close() → 关闭连接

CRUD 示例

插入

import pymysql

conn = pymysql.connect(
    host='127.0.0.1', port=3306,
    user='guest', password='Guest.618',
    database='hrs', charset='utf8mb4'
)
try:
    with conn.cursor() as cursor:
        affected = cursor.execute(
            'INSERT INTO tb_dept VALUES (%s, %s, %s)',
            (no, name, location)
        )
        if affected == 1:
            print('新增成功')
    conn.commit()
except pymysql.MySQLError as err:
    conn.rollback()
    print(err)
finally:
    conn.close()

批量插入（executemany）

data = [(10, '研发部', '成都'), (20, '销售部', '重庆')]
cursor.executemany(
    'INSERT INTO tb_dept VALUES (%s, %s, %s)', data
)

查询

with conn.cursor(pymysql.cursors.DictCursor) as cursor:  # 返回字典
    cursor.execute('SELECT * FROM tb_dept WHERE dno = %s', (no,))
    result = cursor.fetchone()     # 一条
    # result = cursor.fetchall()   # 全部
    # result = cursor.fetchmany(3) # n条

更新 / 删除

with conn.cursor() as cursor:
    affected = cursor.execute(
        'UPDATE tb_dept SET dname = %s WHERE dno = %s',
        (name, no)
    )
    conn.commit()

连接参数

SQL 注入防护

⚠️ 始终使用参数化查询，不要拼接字符串！

# ✅ 正确
cursor.execute('SELECT * FROM tb WHERE id = %s', (id,))

# ❌ 危险！SQL注入
cursor.execute(f'SELECT * FROM tb WHERE id = {id}')

上下文管理器（推荐写法）

with pymysql.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute(...)
        conn.commit()
# 自动 close，无需 finally

连接池（生产环境推荐）

from dbutils.pooled_db import PooledDB

pool = PooledDB(
    creator=pymysql, maxconnections=10,
    host='127.0.0.1', port=3306,
    user='guest', password='Guest.618',
    database='hrs', charset='utf8mb4'
)

conn = pool.connection()  # 从池中取连接
with conn.cursor() as cursor:
    cursor.execute('SELECT 1')
conn.close()  # 归还连接，不关闭

总结

• pymysql.connect() → 游标 cursor() → execute() → commit()
• 参数化查询：用 %s 占位 + 元组传参，防 SQL 注入
• 生产环境：使用 dbutils.PooledDB 连接池