據(jù)庫：從 connect 到 commit 的完整操作指南)
1. 為什么你的 pymysql 一跑就報(bào)錯(cuò)先看清 connect 到 commit 的鏈路很多 Python 開發(fā)者第一次用 pymysql 連數(shù)據(jù)庫代碼看著沒問題運(yùn)行起來卻報(bào)Access denied、Unknown database或者更隱蔽的——數(shù)據(jù)明明插進(jìn)去了重新查詢卻什么都沒有。問題往往不在 SQL 本身而在于沒搞清楚 pymysql 的四個(gè)核心環(huán)節(jié)connect 建立連接、cursor 創(chuàng)建游標(biāo)、execute 執(zhí)行語句、commit 提交事務(wù)。pymysql 是純 Python 實(shí)現(xiàn)的 MySQL 客戶端庫不需要編譯 C 擴(kuò)展pip install pymysql就能用。它適合誰適合做數(shù)據(jù)采集落庫、后臺(tái)腳本、Flask/Django 之外的輕量數(shù)據(jù)操作以及需要快速驗(yàn)證 SQL 邏輯的場景。你可以把它理解成一座橋Python 程序在橋這頭MySQL 服務(wù)在橋那頭connect 是搭橋cursor 是橋上的傳送帶execute 是把貨物放上傳送帶commit 則是按下“確認(rèn)收貨”——不按貨就一直在半路。我試過在本地起一個(gè) MySQL 8.0用 pymysql 走一遍建表、插入、查詢、更新、刪除的完整閉環(huán)中間踩過字符集和事務(wù)的坑。下面把可復(fù)制的配置骨架和驗(yàn)證動(dòng)作拆開講你跟著敲一遍就能獨(dú)立跑通。2. 前置準(zhǔn)備裝庫、建庫、確認(rèn)連接四要素在寫代碼之前先把環(huán)境理清楚。pymysql 的安裝很簡單pip install pymysql如果你用的是虛擬環(huán)境先激活再裝。裝完可以用pip show pymysql確認(rèn)版本一般 1.1.x 就夠用。接著確認(rèn) MySQL 服務(wù)在跑并且你有一個(gè)可用的庫。這里假設(shè)你已經(jīng)建好一個(gè)名為ss的數(shù)據(jù)庫。如果沒有用命令行或客戶端執(zhí)行CREATE DATABASE ss DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;字符集建議統(tǒng)一用utf8mb4否則中文和 emoji 容易出問題。連接四要素必須寫對host、user、password、database。密碼字段在 pymysql 里既可以寫password也可以寫passwd兩者等價(jià)但別拼錯(cuò)。端口默認(rèn) 3306如果改過要顯式指定port。注意連接信息不要硬編碼在業(yè)務(wù)代碼里正式項(xiàng)目建議走環(huán)境變量或配置文件。本文為了演示直觀直接寫在代碼中。3. 可復(fù)制配置connect、cursor、execute、commit 四步骨架先給一個(gè)最小可運(yùn)行骨架把四個(gè)環(huán)節(jié)串起來。這段代碼的作用是連接數(shù)據(jù)庫、查詢 MySQL 版本驗(yàn)證鏈路是否通import pymysql # 第一步connect 建立連接 conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, port3306, charsetutf8mb4 ) try: # 第二步cursor 創(chuàng)建游標(biāo) cursor conn.cursor() # 第三步execute 執(zhí)行 SQL cursor.execute(SELECT VERSION()) # 取回單條結(jié)果 data cursor.fetchone() print(MySQL 版本:, data) finally: # 第四步關(guān)閉游標(biāo)和連接 cursor.close() conn.close()運(yùn)行后如果打印出類似MySQL 版本: (8.0.36,)說明 connect、cursor、execute 三個(gè)環(huán)節(jié)都通了。這里沒有寫 commit因?yàn)椴樵儾桓淖償?shù)據(jù)狀態(tài)不需要提交。接下來建一張表把 commit 加進(jìn)來。表名和字段名建議用反引號(hào)包裹避免和關(guān)鍵字沖突import pymysql conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) try: cursor conn.cursor() sql CREATE TABLE IF NOT EXISTS tast ( first_name VARCHAR(10) PRIMARY KEY, last_name VARCHAR(10), age INT(5), sex VARCHAR(6), income INT(30) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 cursor.execute(sql) # 關(guān)鍵DDL 之后也要 commit否則部分場景下不生效 conn.commit() print(建表成功) except Exception as e: conn.rollback() print(建表失敗:, e) finally: cursor.close() conn.close()這里有個(gè)容易忽略的點(diǎn)CREATE TABLE屬于 DDL在 MySQL 里默認(rèn)是自動(dòng)提交的但顯式寫conn.commit()更穩(wěn)妥也養(yǎng)成習(xí)慣。異常捕獲里加conn.rollback()出錯(cuò)時(shí)回滾避免留下半成品狀態(tài)。4. 增刪改查全流程executemany 批量插入與事務(wù)提交驗(yàn)證建好表后先插入數(shù)據(jù)。單條插入用execute多條插入用executemany后者效率更高import pymysql conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) try: cursor conn.cursor() sql INSERT INTO tast (first_name, last_name, age, sex, income) VALUES (%s, %s, %s, %s, %s) params ( (a, joe, 21, m, 5000), (b, jae, 22, f, 5000), (c, jbe, 23, f, 5000), (d, jvce, 24, m, 5000), ) rows cursor.executemany(sql, params) conn.commit() print(影響行數(shù):, rows) except Exception as e: conn.rollback() print(插入失敗:, e) finally: cursor.close() conn.close()executemany返回的是受影響的行數(shù)這里應(yīng)該是 4。注意參數(shù)用%s占位不要用字符串拼接否則會(huì)有 SQL 注入風(fēng)險(xiǎn)。插入完成后驗(yàn)證查詢import pymysql conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) try: cursor conn.cursor() cursor.execute(SELECT first_name, age, income FROM tast WHERE sex %s, (f,)) results cursor.fetchall() for row in results: print(row) finally: cursor.close() conn.close()fetchall()返回所有結(jié)果fetchone()返回單條。如果查詢結(jié)果為空fetchone()返回None別直接下標(biāo)取值。更新和刪除同樣遵循 execute commit 的模式# 更新 cursor.execute(UPDATE tast SET income %s WHERE first_name %s, (8000, a)) conn.commit() # 刪除 cursor.execute(DELETE FROM tast WHERE first_name %s, (d,)) conn.commit()每次改動(dòng)數(shù)據(jù)后都要 commit否則當(dāng)前連接關(guān)閉時(shí)改動(dòng)會(huì)丟失。你可以做個(gè)實(shí)驗(yàn)把 commit 注釋掉插入數(shù)據(jù)后關(guān)閉連接再重新查詢會(huì)發(fā)現(xiàn)數(shù)據(jù)沒進(jìn)去——這就是事務(wù)未提交的典型表現(xiàn)。5. 驗(yàn)證請求與成功結(jié)果用一條查詢確認(rèn)閉環(huán)完整的驗(yàn)證動(dòng)作是這樣插入數(shù)據(jù)后不關(guān)閉連接直接在同一連接內(nèi)查詢確認(rèn)數(shù)據(jù)可見然后關(guān)閉連接重新建立連接再查一次確認(rèn)數(shù)據(jù)已持久化。import pymysql def get_conn(): return pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) # 第一次連接插入并查詢 conn get_conn() cursor conn.cursor() cursor.execute(INSERT INTO tast VALUES (e, test, 30, m, 9000)) conn.commit() cursor.execute(SELECT COUNT(*) FROM tast) print(當(dāng)前連接內(nèi)行數(shù):, cursor.fetchone()) cursor.close() conn.close() # 第二次連接重新查詢驗(yàn)證持久化 conn2 get_conn() cursor2 conn2.cursor() cursor2.execute(SELECT COUNT(*) FROM tast) print(新連接行數(shù):, cursor2.fetchone()) cursor2.close() conn2.close()兩次打印的行數(shù)一致說明 commit 生效、數(shù)據(jù)已落盤。如果第二次查不到八成是 commit 漏了或者連接用了自動(dòng)提交關(guān)閉的配置。6. 本篇常見錯(cuò)排查從報(bào)錯(cuò)信息定位到具體環(huán)節(jié)下面這張表把高頻報(bào)錯(cuò)和對應(yīng)環(huán)節(jié)列出來方便你對號(hào)入座報(bào)錯(cuò)信息可能原因排查動(dòng)作Access denied for user用戶名或密碼錯(cuò)核對 connect 的 user/passwordUnknown database ss庫不存在或名字拼錯(cuò)執(zhí)行SHOW DATABASES確認(rèn)Cant connect to MySQL server服務(wù)沒起或 host/port 錯(cuò)檢查 MySQL 進(jìn)程和端口插入后查不到數(shù)據(jù)漏了 commit在 execute 后加 conn.commit()中文亂碼字符集不統(tǒng)一connect 加 charsetutf8mb4建庫建表也用 utf8mb4Commands out of sync上一個(gè)結(jié)果集沒讀完用 fetchall 讀完或換新 cursor還有一個(gè)隱蔽的坑cursor和conn忘記關(guān)閉長時(shí)間運(yùn)行會(huì)耗盡連接數(shù)。建議用try/finally或上下文管理器。pymysql 的連接對象支持with語法但游標(biāo)需要手動(dòng)管理寫法如下with pymysql.connect(hostlocalhost, userroot, password123456, databasess, charsetutf8mb4) as conn: with conn.cursor() as cursor: cursor.execute(SELECT 1) print(cursor.fetchone())with塊結(jié)束時(shí)會(huì)自動(dòng) commit如果沒異?；?rollback如果有異常并關(guān)閉連接。這個(gè)寫法能省掉不少樣板代碼。如果你在排查連接或權(quán)限問題時(shí)需要更細(xì)的日志可以在 connect 里加cursorclasspymysql.cursors.DictCursor讓結(jié)果以字典返回字段名一目了然調(diào)試時(shí)更直觀。7. 把連接配置沉淀成可復(fù)用模塊跑通一次閉環(huán)后建議把連接邏輯抽成函數(shù)或類避免每次重復(fù)寫四要素。一個(gè)簡單的做法是用字典存配置從環(huán)境變量讀取import os import pymysql def get_connection(): return pymysql.connect( hostos.getenv(DB_HOST, localhost), useros.getenv(DB_USER, root), passwordos.getenv(DB_PASSWORD, 123456), databaseos.getenv(DB_NAME, ss), charsetutf8mb4, cursorclasspymysql.cursors.DictCursor )這樣本地開發(fā)和線上部署只需改環(huán)境變量代碼不用動(dòng)。配合with語法一個(gè)完整的讀寫操作可以壓縮到幾行with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(SELECT * FROM tast WHERE age %s, (20,)) for row in cursor.fetchall(): print(row[first_name], row[income])到這里connect、cursor、execute、commit 四個(gè)環(huán)節(jié)你已經(jīng)完整走了一遍。接下來可以試著把 SQL 換成自己的業(yè)務(wù)表或者接入定時(shí)任務(wù)做批量落庫。如果后續(xù)要接大模型做數(shù)據(jù)問答、或者把數(shù)據(jù)庫操作封裝成 Agent 工具可以到 TaoToken 的模型對話頁面先驗(yàn)證模型對 SQL 的理解能力再?zèng)Q定怎么把 pymysql 的讀寫鏈路嵌進(jìn)你的編碼工作流。 用 pymysql 連接數(shù)據(jù)庫：從 connect 到 commit 的完整操作指南
1. 為什么你的 pymysql 一跑就報(bào)錯(cuò)先看清 connect 到 commit 的鏈路很多 Python 開發(fā)者第一次用 pymysql 連數(shù)據(jù)庫代碼看著沒問題運(yùn)行起來卻報(bào)Access denied、Unknown database或者更隱蔽的——數(shù)據(jù)明明插進(jìn)去了重新查詢卻什么都沒有。問題往往不在 SQL 本身而在于沒搞清楚 pymysql 的四個(gè)核心環(huán)節(jié)connect 建立連接、cursor 創(chuàng)建游標(biāo)、execute 執(zhí)行語句、commit 提交事務(wù)。pymysql 是純 Python 實(shí)現(xiàn)的 MySQL 客戶端庫不需要編譯 C 擴(kuò)展pip install pymysql就能用。它適合誰適合做數(shù)據(jù)采集落庫、后臺(tái)腳本、Flask/Django 之外的輕量數(shù)據(jù)操作以及需要快速驗(yàn)證 SQL 邏輯的場景。你可以把它理解成一座橋Python 程序在橋這頭MySQL 服務(wù)在橋那頭connect 是搭橋cursor 是橋上的傳送帶execute 是把貨物放上傳送帶commit 則是按下“確認(rèn)收貨”——不按貨就一直在半路。我試過在本地起一個(gè) MySQL 8.0用 pymysql 走一遍建表、插入、查詢、更新、刪除的完整閉環(huán)中間踩過字符集和事務(wù)的坑。下面把可復(fù)制的配置骨架和驗(yàn)證動(dòng)作拆開講你跟著敲一遍就能獨(dú)立跑通。2. 前置準(zhǔn)備裝庫、建庫、確認(rèn)連接四要素在寫代碼之前先把環(huán)境理清楚。pymysql 的安裝很簡單pip install pymysql如果你用的是虛擬環(huán)境先激活再裝。裝完可以用pip show pymysql確認(rèn)版本一般 1.1.x 就夠用。接著確認(rèn) MySQL 服務(wù)在跑并且你有一個(gè)可用的庫。這里假設(shè)你已經(jīng)建好一個(gè)名為ss的數(shù)據(jù)庫。如果沒有用命令行或客戶端執(zhí)行CREATE DATABASE ss DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;字符集建議統(tǒng)一用utf8mb4否則中文和 emoji 容易出問題。連接四要素必須寫對host、user、password、database。密碼字段在 pymysql 里既可以寫password也可以寫passwd兩者等價(jià)但別拼錯(cuò)。端口默認(rèn) 3306如果改過要顯式指定port。注意連接信息不要硬編碼在業(yè)務(wù)代碼里正式項(xiàng)目建議走環(huán)境變量或配置文件。本文為了演示直觀直接寫在代碼中。3. 可復(fù)制配置connect、cursor、execute、commit 四步骨架先給一個(gè)最小可運(yùn)行骨架把四個(gè)環(huán)節(jié)串起來。這段代碼的作用是連接數(shù)據(jù)庫、查詢 MySQL 版本驗(yàn)證鏈路是否通import pymysql # 第一步connect 建立連接 conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, port3306, charsetutf8mb4 ) try: # 第二步cursor 創(chuàng)建游標(biāo) cursor conn.cursor() # 第三步execute 執(zhí)行 SQL cursor.execute(SELECT VERSION()) # 取回單條結(jié)果 data cursor.fetchone() print(MySQL 版本:, data) finally: # 第四步關(guān)閉游標(biāo)和連接 cursor.close() conn.close()運(yùn)行后如果打印出類似MySQL 版本: (8.0.36,)說明 connect、cursor、execute 三個(gè)環(huán)節(jié)都通了。這里沒有寫 commit因?yàn)椴樵儾桓淖償?shù)據(jù)狀態(tài)不需要提交。接下來建一張表把 commit 加進(jìn)來。表名和字段名建議用反引號(hào)包裹避免和關(guān)鍵字沖突import pymysql conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) try: cursor conn.cursor() sql CREATE TABLE IF NOT EXISTS tast ( first_name VARCHAR(10) PRIMARY KEY, last_name VARCHAR(10), age INT(5), sex VARCHAR(6), income INT(30) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 cursor.execute(sql) # 關(guān)鍵DDL 之后也要 commit否則部分場景下不生效 conn.commit() print(建表成功) except Exception as e: conn.rollback() print(建表失敗:, e) finally: cursor.close() conn.close()這里有個(gè)容易忽略的點(diǎn)CREATE TABLE屬于 DDL在 MySQL 里默認(rèn)是自動(dòng)提交的但顯式寫conn.commit()更穩(wěn)妥也養(yǎng)成習(xí)慣。異常捕獲里加conn.rollback()出錯(cuò)時(shí)回滾避免留下半成品狀態(tài)。4. 增刪改查全流程executemany 批量插入與事務(wù)提交驗(yàn)證建好表后先插入數(shù)據(jù)。單條插入用execute多條插入用executemany后者效率更高import pymysql conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) try: cursor conn.cursor() sql INSERT INTO tast (first_name, last_name, age, sex, income) VALUES (%s, %s, %s, %s, %s) params ( (a, joe, 21, m, 5000), (b, jae, 22, f, 5000), (c, jbe, 23, f, 5000), (d, jvce, 24, m, 5000), ) rows cursor.executemany(sql, params) conn.commit() print(影響行數(shù):, rows) except Exception as e: conn.rollback() print(插入失敗:, e) finally: cursor.close() conn.close()executemany返回的是受影響的行數(shù)這里應(yīng)該是 4。注意參數(shù)用%s占位不要用字符串拼接否則會(huì)有 SQL 注入風(fēng)險(xiǎn)。插入完成后驗(yàn)證查詢import pymysql conn pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) try: cursor conn.cursor() cursor.execute(SELECT first_name, age, income FROM tast WHERE sex %s, (f,)) results cursor.fetchall() for row in results: print(row) finally: cursor.close() conn.close()fetchall()返回所有結(jié)果fetchone()返回單條。如果查詢結(jié)果為空fetchone()返回None別直接下標(biāo)取值。更新和刪除同樣遵循 execute commit 的模式# 更新 cursor.execute(UPDATE tast SET income %s WHERE first_name %s, (8000, a)) conn.commit() # 刪除 cursor.execute(DELETE FROM tast WHERE first_name %s, (d,)) conn.commit()每次改動(dòng)數(shù)據(jù)后都要 commit否則當(dāng)前連接關(guān)閉時(shí)改動(dòng)會(huì)丟失。你可以做個(gè)實(shí)驗(yàn)把 commit 注釋掉插入數(shù)據(jù)后關(guān)閉連接再重新查詢會(huì)發(fā)現(xiàn)數(shù)據(jù)沒進(jìn)去——這就是事務(wù)未提交的典型表現(xiàn)。5. 驗(yàn)證請求與成功結(jié)果用一條查詢確認(rèn)閉環(huán)完整的驗(yàn)證動(dòng)作是這樣插入數(shù)據(jù)后不關(guān)閉連接直接在同一連接內(nèi)查詢確認(rèn)數(shù)據(jù)可見然后關(guān)閉連接重新建立連接再查一次確認(rèn)數(shù)據(jù)已持久化。import pymysql def get_conn(): return pymysql.connect( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) # 第一次連接插入并查詢 conn get_conn() cursor conn.cursor() cursor.execute(INSERT INTO tast VALUES (e, test, 30, m, 9000)) conn.commit() cursor.execute(SELECT COUNT(*) FROM tast) print(當(dāng)前連接內(nèi)行數(shù):, cursor.fetchone()) cursor.close() conn.close() # 第二次連接重新查詢驗(yàn)證持久化 conn2 get_conn() cursor2 conn2.cursor() cursor2.execute(SELECT COUNT(*) FROM tast) print(新連接行數(shù):, cursor2.fetchone()) cursor2.close() conn2.close()兩次打印的行數(shù)一致說明 commit 生效、數(shù)據(jù)已落盤。如果第二次查不到八成是 commit 漏了或者連接用了自動(dòng)提交關(guān)閉的配置。6. 本篇常見錯(cuò)排查從報(bào)錯(cuò)信息定位到具體環(huán)節(jié)下面這張表把高頻報(bào)錯(cuò)和對應(yīng)環(huán)節(jié)列出來方便你對號(hào)入座報(bào)錯(cuò)信息可能原因排查動(dòng)作Access denied for user用戶名或密碼錯(cuò)核對 connect 的 user/passwordUnknown database ss庫不存在或名字拼錯(cuò)執(zhí)行SHOW DATABASES確認(rèn)Cant connect to MySQL server服務(wù)沒起或 host/port 錯(cuò)檢查 MySQL 進(jìn)程和端口插入后查不到數(shù)據(jù)漏了 commit在 execute 后加 conn.commit()中文亂碼字符集不統(tǒng)一connect 加 charsetutf8mb4建庫建表也用 utf8mb4Commands out of sync上一個(gè)結(jié)果集沒讀完用 fetchall 讀完或換新 cursor還有一個(gè)隱蔽的坑cursor和conn忘記關(guān)閉長時(shí)間運(yùn)行會(huì)耗盡連接數(shù)。建議用try/finally或上下文管理器。pymysql 的連接對象支持with語法但游標(biāo)需要手動(dòng)管理寫法如下with pymysql.connect(hostlocalhost, userroot, password123456, databasess, charsetutf8mb4) as conn: with conn.cursor() as cursor: cursor.execute(SELECT 1) print(cursor.fetchone())with塊結(jié)束時(shí)會(huì)自動(dòng) commit如果沒異?；?rollback如果有異常并關(guān)閉連接。這個(gè)寫法能省掉不少樣板代碼。如果你在排查連接或權(quán)限問題時(shí)需要更細(xì)的日志可以在 connect 里加cursorclasspymysql.cursors.DictCursor讓結(jié)果以字典返回字段名一目了然調(diào)試時(shí)更直觀。7. 把連接配置沉淀成可復(fù)用模塊跑通一次閉環(huán)后建議把連接邏輯抽成函數(shù)或類避免每次重復(fù)寫四要素。一個(gè)簡單的做法是用字典存配置從環(huán)境變量讀取import os import pymysql def get_connection(): return pymysql.connect( hostos.getenv(DB_HOST, localhost), useros.getenv(DB_USER, root), passwordos.getenv(DB_PASSWORD, 123456), databaseos.getenv(DB_NAME, ss), charsetutf8mb4, cursorclasspymysql.cursors.DictCursor )這樣本地開發(fā)和線上部署只需改環(huán)境變量代碼不用動(dòng)。配合with語法一個(gè)完整的讀寫操作可以壓縮到幾行with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(SELECT * FROM tast WHERE age %s, (20,)) for row in cursor.fetchall(): print(row[first_name], row[income])到這里connect、cursor、execute、commit 四個(gè)環(huán)節(jié)你已經(jīng)完整走了一遍。接下來可以試著把 SQL 換成自己的業(yè)務(wù)表或者接入定時(shí)任務(wù)做批量落庫。如果后續(xù)要接大模型做數(shù)據(jù)問答、或者把數(shù)據(jù)庫操作封裝成 Agent 工具可以到 TaoToken 的模型對話頁面先驗(yàn)證模型對 SQL 的理解能力再?zèng)Q定怎么把 pymysql 的讀寫鏈路嵌進(jìn)你的編碼工作流。