如何在Python中操作MySQL？

平时写一些后台脚本、数据清洗任务，或者做接口服务，Python 连 MySQL 基本绕不过去。刚开始大家通常是“能连上就行”，后面一上业务，问题就慢慢出来了：连接没关、事务没提交、批量插入太慢、查询结果拿回来不好处理。

这篇我不铺太大，就按平时开发里最常见的几件事来写：连接、查询、插入、事务、批量操作。代码我用 pymysql，够轻，也比较常见。参考你给我的技术文风格，我尽量按“代码 + 场景”往下走。

先安装依赖：

pip install pymysql

最先要有一个能正常复用的连接。很多人一上来直接 connect()，查完就不管了，脚本短一点问题不大，服务跑久了就容易把连接占着。

import pymysql

defget_conn():
return pymysql.connect(
        host="127.0.0.1",
        port=3306,
        user="app_user",
        password="123456",
        database="demo_db",
        charset="utf8mb4",
        autocommit=False,
        cursorclass=pymysql.cursors.DictCursor
    )

这里我一般会把 cursorclass 直接设成 DictCursor，这样查出来是字典，后面写业务判断顺手一点，不用再去记下标位置。

比如查一个用户：

deffind_user(user_id: int):
    conn = get_conn()
try:
with conn.cursor() as cursor:
            sql = """
            select id, username, status, created_at
            from user_account
            where id = %s
            """
            cursor.execute(sql, (user_id,))
return cursor.fetchone()
finally:
        conn.close()

调用后拿到的数据大概就是这样：

{'id': 12, 'username': 'dongge', 'status': 1, 'created_at': datetime.datetime(...)}

这类单条查询不复杂，真正要注意的是 SQL 参数别自己拼字符串。下面这种我不建议：

sql = f"select * from user_account where id = {user_id}"

不是说一定会出事，而是后面条件一多，模糊查询、分页、动态排序一混进去，就容易埋雷。参数化写法虽然啰嗦一点，但值和 SQL 结构是分开的，稳一些。

再看插入。插入最容易漏的是 commit()。

defadd_user(username: str, status: int):
    conn = get_conn()
try:
with conn.cursor() as cursor:
            sql = """
            insert into user_account(username, status, created_at)
            values(%s, %s, now())
            """
            cursor.execute(sql, (username, status))
        conn.commit()
returnTrue
except Exception:
        conn.rollback()
raise
finally:
        conn.close()

这里的习惯我建议固定下来：写操作一律 try / except / rollback / commit。 别觉得脚本简单就省，后面一旦不是单表写入，而是“插用户 + 记日志 + 扣额度”这种连续动作，不带事务基本就是给自己留坑。

举个稍微接近业务一点的例子，注册用户的时候顺手写一条初始化记录：

defregister_user(username: str):
    conn = get_conn()
try:
with conn.cursor() as cursor:
            cursor.execute(
"insert into user_account(username, status, created_at) values(%s, 1, now())",
                (username,)
            )
            user_id = cursor.lastrowid

            cursor.execute(
"insert into user_profile(user_id, nickname, level) values(%s, %s, %s)",
                (user_id, username, 1)
            )

        conn.commit()
return user_id
except Exception:
        conn.rollback()
raise
finally:
        conn.close()

这种代码如果没有事务，第一条成功第二条失败，库里就会留下半截数据。现场查起来其实挺烦，尤其是业务方只会告诉你一句：“这个用户看起来注册成功了，但又不完整。”

更新也差不多，重点是看影响行数。别 update 发出去了就当它成功了。

defdisable_user(user_id: int):
    conn = get_conn()
try:
with conn.cursor() as cursor:
            sql = """
            update user_account
            set status = 0, updated_at = now()
            where id = %s and status = 1
            """
            affected = cursor.execute(sql, (user_id,))
        conn.commit()
return affected
except Exception:
        conn.rollback()
raise
finally:
        conn.close()

affected 如果是 0，通常就要多看一眼：到底是这个用户不存在，还是状态本来就不是 1。线上排查时，这个返回值比一句“更新成功”有用得多。参考你给的故障类文章，真正推进判断的，往往就是这种很小的执行结果。

再说批量插入。很多人第一次导数据时，喜欢一条一条 execute()，几千条还能忍，几万条就开始慢了。

defbatch_add_orders(order_list: list[tuple]):
    conn = get_conn()
try:
with conn.cursor() as cursor:
            sql = """
            insert into order_record(order_no, user_id, amount, created_at)
            values(%s, %s, %s, now())
            """
            cursor.executemany(sql, order_list)
        conn.commit()
except Exception:
        conn.rollback()
raise
finally:
        conn.close()

调用时这样传：

orders = [
    ("A20260309001", 101, 88.50),
    ("A20260309002", 102, 19.90),
    ("A20260309003", 103, 256.00),
]
batch_add_orders(orders)

executemany() 不一定能解决所有性能问题，但比单条循环插入通常会好不少。 如果数据量再大，就别一次塞十几万条，分批处理更稳一点，比如一批 500 或 1000。

查询结果很多时，分页也得顺手带上。这个写法在后台管理里很常见：

defquery_users(page: int, page_size: int):
    offset = (page - 1) * page_size
    conn = get_conn()
try:
with conn.cursor() as cursor:
            sql = """
            select id, username, status
            from user_account
            order by id desc
            limit %s, %s
            """
            cursor.execute(sql, (offset, page_size))
return cursor.fetchall()
finally:
        conn.close()

这里有个小点，limit 也照样走参数，不要自己拼。统一风格后，后面查问题时一眼就知道这段 SQL 的变量入口在哪。

最后补一个我平时比较常用的封装思路。不是做成多完整的 ORM，而是把“执行 SQL”这件事收一下，脚本里少写重复代码。

classMySQLClient:
def__init__(self):
        self.conn = get_conn()

defquery_one(self, sql, params=None):
with self.conn.cursor() as cursor:
            cursor.execute(sql, params or ())
return cursor.fetchone()

defquery_all(self, sql, params=None):
with self.conn.cursor() as cursor:
            cursor.execute(sql, params or ())
return cursor.fetchall()

defexecute(self, sql, params=None):
with self.conn.cursor() as cursor:
            rows = cursor.execute(sql, params or ())
        self.conn.commit()
return rows

defclose(self):
        self.conn.close()

用起来就会简单一些：

db = MySQLClient()
user = db.query_one("select id, username from user_account where id = %s", (1,))
db.close()

这种封装不重，但够用。尤其写内部工具、定时任务、小管理后台时，比每次都把连接和游标逻辑摊开要省事不少。

Python 操作 MySQL，本质上不复杂，难点不在“怎么连”，而在你是不是把几个基本动作做扎实了：查询走参数化、写操作带事务、批量数据别一条条插、连接用完及时关。