pip install mysql-connector-python

pip install pymysql

# Windows
pip install mysqlclient
# Linux (Ubuntu/Debian)
sudo apt-get install python3-dev default-libmysqlclient-dev build-essential
pip install mysqlclient
# CentOS/RHEL
sudo yum install python3-devel mysql-devel
pip install mysqlclient

import pymysql

# 建立數(shù)據(jù)庫連接
db = pymysql.connect(
    host="localhost",
    user="root",
    password="your_password",
    database="testdb",
    port=3306,
    charset="utf8mb4",
    connect_timeout=10
)

cursor = db.cursor()

from pymysql.cursors import DictCursor
cursor = db.cursor(cursor=DictCursor)

# 查詢示例
cursor.execute("SELECT * FROM users")

# 獲取所有結(jié)果
results = cursor.fetchall()
for row in results:
    print(row)

# 獲取單條結(jié)果
one_row = cursor.fetchone()
print(one_row)

cursor.close()
db.close()

# 連接上下文管理器
with pymysql.connect(
    host="localhost",
    user="root",
    password="your_password",
    database="testdb",
    charset="utf8mb4"
) as db:
    # 游標(biāo)上下文管理器
    with db.cursor() as cursor:
        cursor.execute("SELECT * FROM users")
        results = cursor.fetchall()
        for row in results:
            print(row)
# 自動關(guān)閉連接和游標(biāo)

# 單條插入
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
cursor.execute(sql, ("Alice", 25))
db.commit()  # 提交事務(wù)，保存數(shù)據(jù)

# 批量插入（executemany）
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
data = [("Bob", 30), ("Charlie", 35), ("David", 28)]
cursor.executemany(sql, data)
db.commit()

sql = "UPDATE users SET age = %s WHERE name = %s"
cursor.execute(sql, (26, "Alice"))
db.commit()

sql = "DELETE FROM users WHERE name = %s"
cursor.execute(sql, ("Alice",))  # 注意逗號，確保是元組
db.commit()

cursor.execute("SELECT * FROM users")
results = cursor.fetchall()

sql = "SELECT * FROM users WHERE age > %s"
cursor.execute(sql, (25,))
results = cursor.fetchall()

# 查詢名字包含"a"的用戶
sql = "SELECT * FROM users WHERE name LIKE %s"
cursor.execute(sql, ("%a%",))
results = cursor.fetchall()

sql = """
SELECT users.name, orders.amount
FROM users
INNER JOIN orders ON users.id = orders.user_id
"""
cursor.execute(sql)
results = cursor.fetchall()

# 危險！用戶輸入惡意內(nèi)容會導(dǎo)致SQL注入
sql = f"SELECT * FROM users WHERE name = '{user_input}'"

sql = "SELECT * FROM users WHERE name = %s"
cursor.execute(sql, (user_input,))

pip install dbutils

from dbutils.pooled_db import PooledDB
import pymysql

# 數(shù)據(jù)庫連接配置
dbconfig = {
    "host": "localhost",
    "user": "root",
    "password": "your_password",
    "database": "testdb",
    "port": 3306,
    "charset": "utf8mb4"
}

# 創(chuàng)建連接池
connection_pool = PooledDB(
    creator=pymysql,  # 使用PyMySQL作為連接庫
    maxconnections=5,  # 最大連接數(shù)，根據(jù)數(shù)據(jù)庫配置調(diào)整
    mincached=2,  # 初始化時創(chuàng)建的空閑連接數(shù)
    maxcached=3,  # 最大空閑連接數(shù)
    maxshared=3,  # 最大共享連接數(shù)
    blocking=True,  # 連接池滿時是否阻塞等待
    setsession=[],  # 會話初始化SQL
    ping=1,  # 自動檢測連接有效性（1=每次獲取連接時檢測）
    **dbconfig
)

# 從連接池獲取連接
db_connection = connection_pool.connection()
cursor = db_connection.cursor()

# 執(zhí)行SQL操作
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
    print(row)

# 關(guān)閉游標(biāo)和連接（連接自動歸還到連接池）
cursor.close()
db_connection.close()

# 關(guān)閉自動提交（開啟事務(wù)模式）
db.autocommit = False
# 或顯式開啟事務(wù)
cursor.execute("START TRANSACTION")

db.commit()

db.rollback()

# 1. READ UNCOMMITTED
cursor.execute("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED")

# 2. READ COMMITTED
cursor.execute("SET TRANSACTION ISOLATION LEVEL READ COMMITTED")

# 3. REPEATABLE READ（MySQL默認(rèn)）
cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")

# 4. SERIALIZABLE
cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")

import pymysql

# 連接數(shù)據(jù)庫
conn = pymysql.connect(
    host="localhost",
    user="root",
    password="your_password",
    database="testdb",
    charset="utf8mb4"
)
cursor = conn.cursor()

# 關(guān)閉自動提交，開啟事務(wù)
conn.autocommit = False

try:
    # 1. 從賬戶A扣款100元
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE name = %s", ("Alice",))
    
    # 2. 向賬戶B入賬100元
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE name = %s", ("Bob",))
    
    # 提交事務(wù)
    conn.commit()
    print("轉(zhuǎn)賬成功")
except pymysql.MySQLError as err:
    # 發(fā)生錯誤，回滾事務(wù)
    print(f"轉(zhuǎn)賬失敗: {err}")
    conn.rollback()
finally:
    # 關(guān)閉游標(biāo)和連接
    cursor.close()
    conn.close()

cursor.execute("SAVEPOINT point1")
# 部分操作
cursor.execute("ROLLBACK TO SAVEPOINT point1")

import pymysql
from pymysql.err import MySQLError, OperationalError, ProgrammingError

try:
    # 數(shù)據(jù)庫操作
    conn = pymysql.connect(...)
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    conn.commit()
except OperationalError as e:
    print(f"連接錯誤: {e}")
    # 重連邏輯
except ProgrammingError as e:
    print(f"SQL語法錯誤: {e}")
except MySQLError as e:
    print(f"數(shù)據(jù)庫錯誤: {e}")
finally:
    if 'cursor' in locals():
        cursor.close()
    if 'conn' in locals():
        conn.close()

sql = """
LOAD DATA LOCAL INFILE 'data.csv'
INTO TABLE users
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
IGNORE 1 ROWS
"""
cursor.execute(sql)
conn.commit()

cursor.execute("EXPLAIN SELECT * FROM users WHERE age > 25")
print(cursor.fetchall())

def get_connection():
    try:
        conn = pymysql.connect(...)
        return conn
    except OperationalError:
        # 重連3次
        for i in range(3):
            try:
                conn = pymysql.connect(...)
                return conn
            except OperationalError:
                time.sleep(1)
        raise Exception("連接數(shù)據(jù)庫失敗")

# 主庫（寫操作）
master_conn = pymysql.connect(host="master_host", ...)
# 從庫（讀操作）
slave_conn = pymysql.connect(host="slave_host", ...)

# 寫操作走主庫
with master_conn.cursor() as cursor:
    cursor.execute("INSERT INTO users VALUES (%s, %s)", ("Eve", 22))
    master_conn.commit()

# 讀操作走從庫
with slave_conn.cursor() as cursor:
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()

import pymysql
from dbutils.pooled_db import PooledDB

# 1. 初始化連接池
dbconfig = {
    "host": "localhost",
    "user": "root",
    "password": "your_password",
    "database": "testdb",
    "port": 3306,
    "charset": "utf8mb4"
}
pool = PooledDB(
    creator=pymysql,
    maxconnections=5,
    mincached=2,
    ping=1,
    **dbconfig
)

# 2. 用戶CRUD操作
class UserManager:
    def __init__(self):
        self.conn = pool.connection()
        self.cursor = self.conn.cursor()
    
    def add_user(self, name, age):
        """添加用戶"""
        try:
            sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
            self.cursor.execute(sql, (name, age))
            self.conn.commit()
            return True
        except pymysql.MySQLError as e:
            print(f"添加用戶失敗: {e}")
            self.conn.rollback()
            return False
    
    def get_user_by_name(self, name):
        """根據(jù)姓名查詢用戶"""
        sql = "SELECT * FROM users WHERE name = %s"
        self.cursor.execute(sql, (name,))
        return self.cursor.fetchone()
    
    def update_user_age(self, name, new_age):
        """更新用戶年齡"""
        try:
            sql = "UPDATE users SET age = %s WHERE name = %s"
            self.cursor.execute(sql, (new_age, name))
            self.conn.commit()
            return True
        except pymysql.MySQLError as e:
            print(f"更新用戶失敗: {e}")
            self.conn.rollback()
            return False
    
    def delete_user(self, name):
        """刪除用戶"""
        try:
            sql = "DELETE FROM users WHERE name = %s"
            self.cursor.execute(sql, (name,))
            self.conn.commit()
            return True
        except pymysql.MySQLError as e:
            print(f"刪除用戶失敗: {e}")
            self.conn.rollback()
            return False
    
    def close(self):
        """關(guān)閉連接（歸還到連接池）"""
        self.cursor.close()
        self.conn.close()

# 3. 測試
if __name__ == "__main__":
    manager = UserManager()
    # 添加用戶
    manager.add_user("Eve", 22)
    # 查詢用戶
    print(manager.get_user_by_name("Eve"))
    # 更新年齡
    manager.update_user_age("Eve", 23)
    # 刪除用戶
    manager.delete_user("Eve")
    # 關(guān)閉連接
    manager.close()