在 Python 中操作 MySQL 數據庫主要通過 pymysql 或 mysql-connector-python 庫實現。以下是完整的技術指南，包含連接管理、CRUD 操作和最佳實踐：
一、環境準備
1. 安裝驅動庫
pip install pymysql # 推薦（純Python實現）
# 或
pip install mysql-connector-python # Oracle官方驅動
2. 數據庫準備
CREATE DATABASE testdb;
USE testdb;CREATE TABLE users (id INT AUTO_INCREMENT PRIMARY KEY,username VARCHAR(50) UNIQUE NOT NULL,email VARCHAR(100) NOT NULL,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
二、連接管理
1. 基礎連接
import pymysql# 建立連接
conn = pymysql.connect(host='localhost',user='root',password='your_password',database='testdb',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor # 返回字典格式結果
)
2. 上下文管理器（推薦）
with pymysql.connect(host='localhost', user='root', password='your_pwd', database='testdb') as conn:with conn.cursor() as cursor:# 執行SQL操作pass# 連接在此處自動提交/回滾（取決于autocommit設置）
三、CRUD 操作
1. 查詢數據
def get_user(username):try:with conn.cursor() as cursor:sql = "SELECT * FROM users WHERE username = %s"cursor.execute(sql, (username,))result = cursor.fetchone() # 獲取單條記錄return resultexcept pymysql.MySQLError as e:print(f"數據庫錯誤: {e}")return Noneuser = get_user("john_doe")
print(user) # 輸出: {'id': 1, 'username': 'john_doe', ...}
2. 插入數據
def create_user(username, email):try:with conn.cursor() as cursor:sql = """INSERT INTO users (username, email) VALUES (%s, %s)"""cursor.execute(sql, (username, email))conn.commit() # 顯式提交事務return cursor.lastrowid # 返回自增IDexcept pymysql.IntegrityError:print("用戶名已存在")conn.rollback()return Nonenew_id = create_user("jane_smith", "jane@example.com")
3. 更新數據
def update_email(user_id, new_email):try:with conn.cursor() as cursor:sql = "UPDATE users SET email = %s WHERE id = %s"affected_rows = cursor.execute(sql, (new_email, user_id))conn.commit()return affected_rows > 0except pymysql.MySQLError:conn.rollback()return False
4. 批量操作
def batch_insert(users):try:with conn.cursor() as cursor:sql = "INSERT INTO users (username, email) VALUES (%s, %s)"cursor.executemany(sql, users) # 批量執行conn.commit()return cursor.rowcountexcept pymysql.MySQLError:conn.rollback()return 0batch_insert([("user1", "u1@test.com"),("user2", "u2@test.com")
])
四、高級技巧
1. 連接池管理
from dbutils.pooled_db import PooledDBpool = PooledDB(creator=pymysql,maxconnections=10,host='localhost',user='root',password='your_pwd',database='testdb'
)# 使用連接池獲取連接
conn = pool.connection()
2. 存儲過程調用
with conn.cursor() as cursor:cursor.callproc('get_user_stats', (1,)) # 調用存儲過程result = cursor.fetchall()
3. 事務控制
try:with conn.cursor() as cursor:# 執行多個操作cursor.execute("UPDATE account SET balance = balance - 100 WHERE id = 1")cursor.execute("UPDATE account SET balance = balance + 100 WHERE id = 2")conn.commit() # 顯式提交事務
except:conn.rollback() # 回滾所有操作
五、安全實踐
1. 防止SQL注入
- 永遠使用參數化查詢（不要拼接字符串）
# 正確方式 cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))# 危險方式（禁用！） cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
2. 敏感信息處理
- 使用環境變量存儲密碼：
import os password = os.getenv('DB_PASSWORD')
3. 連接超時設置
conn = pymysql.connect(connect_timeout=5, # 連接超時5秒read_timeout=10 # 讀取超時10秒
)
六、性能優化
1. 查詢優化
-
添加索引：
ALTER TABLE users ADD INDEX idx_email (email); -
使用EXPLAIN分析查詢：
with conn.cursor() as cursor:cursor.execute("EXPLAIN SELECT * FROM users WHERE username = %s", ("john",))print(cursor.fetchall())
2. 結果集處理
-
分頁查詢：
sql = "SELECT * FROM users LIMIT %s OFFSET %s" cursor.execute(sql, (page_size, (page-1)*page_size)) -
流式讀取（大數據量）：
with conn.cursor(pymysql.cursors.SSCursor) as cursor: # 使用服務器端游標cursor.execute("SELECT * FROM large_table")for row in cursor:process(row)
七、完整示例
import pymysql
from contextlib import contextmanager@contextmanager
def database_connection():conn = pymysql.connect(host='localhost',user='root',password='your_pwd',database='testdb',charset='utf8mb4')try:yield connfinally:conn.close()def main():with database_connection() as conn:with conn.cursor() as cursor:# 創建用戶cursor.execute("""INSERT INTO users (username, email)VALUES (%s, %s)""", ("new_user", "user@example.com"))user_id = cursor.lastrowid# 查詢用戶cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))print(cursor.fetchone())# 更新記錄cursor.execute("""UPDATE users SET email = %s WHERE id = %s""", ("new_email@example.com", user_id))conn.commit()if __name__ == "__main__":main()
八、故障排查
1. 常見錯誤碼
1045: 訪問被拒絕（檢查用戶名/密碼）2003: 無法連接（檢查主機/端口）1062: 唯一鍵沖突1146: 表不存在
2. 日志記錄
import logging
logging.basicConfig(level=logging.DEBUG,format='%(asctime)s - %(levelname)s - %(message)s'
)# 在連接參數中添加
conn = pymysql.connect(..., cursorclass=pymysql.cursors.SSDictCursor,client_flag=pymysql.client.CLIENT.MULTI_STATEMENTS)
通過遵循這些實踐，可以構建安全、高效的數據庫交互應用。對于復雜場景，建議結合ORM框架（如SQLAlchemy）進行抽象層開發。
)
