pip install pymysql

import pymysql
from pymysql.cursors import DictCursor

# 推薦配置方式
def create_connection():
    return pymysql.connect(
        host='localhost',      # 數(shù)據(jù)庫地址
        user='username',       # 用戶名
        password='password',   # 密碼
        database='test_db',    # 數(shù)據(jù)庫名
        port=3306,            # 端口，默認(rèn)3306
        charset='utf8mb4',     # 字符集，推薦utf8mb4
        autocommit=False,     # 是否自動提交
        cursorclass=DictCursor # 返回字典格式結(jié)果
    )
	

import pymysql
from pymysql.cursors import DictCursor

def get_db_connection():
    """獲取數(shù)據(jù)庫連接"""
    return pymysql.connect(
        host='localhost',
        user='myuser',
        password='mypassword',
        database='mydatabase',
        charset='utf8mb4',
        autocommit=False,
        cursorclass=DictCursor,
        connect_timeout=10  # 連接超時10秒
    )

# 使用示例
def test_connection():
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT 1 as test")
            result = cursor.fetchone()
            print("連接測試成功:", result)
    finally:
        conn.close()

test_connection()

連接測試成功: {'test': 1}

CREATE TABLE mydb.users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    age INT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

import pymysql
from pymysql.cursors import DictCursor
from typing import List, Dict, Any, Optional, Tuple

class MySQLManager:
    """MySQL 數(shù)據(jù)庫管理類"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config

    def execute_query(self, sql: str, params: Tuple = None) -> List[Dict]:
        """執(zhí)行查詢語句（SELECT）"""
        conn = pymysql.connect(**self.config)
        try:
            with conn.cursor(DictCursor) as cursor:
                cursor.execute(sql, params or ())
                return cursor.fetchall()
        finally:
            conn.close()

    def execute_update(self, sql: str, params: Tuple = None) -> int:
        """執(zhí)行更新語句（INSERT/UPDATE/DELETE）"""
        conn = pymysql.connect(**self.config)
        try:
            with conn.cursor() as cursor:
                affected_rows = cursor.execute(sql, params or ())
                conn.commit()
                return affected_rows
        except Exception as e:
            conn.rollback()
            raise e
        finally:
            conn.close()			

# 數(shù)據(jù)庫配置
db_config = {
    'host': 'localhost',
    'user': 'root',
    'password': 'password',
    'database': 'test_db',
    'charset': 'utf8mb4',
    'cursorclass': DictCursor 
}

db = MySQLManager(db_config)

# 1. 插入數(shù)據(jù)
def add_user(name: str, email: str, age: int) -> int:
    sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
    return db.execute_update(sql, (name, email, age))

# 2. 查詢數(shù)據(jù)
def get_all_users() -> List[Dict]:
    return db.execute_query("SELECT * FROM users")

# 3. 更新數(shù)據(jù)
def update_user_email(user_id: int, new_email: str) -> int:
    sql = "UPDATE users SET email = %s WHERE id = %s"
    return db.execute_update(sql, (new_email, user_id))

# 4. 刪除數(shù)據(jù)
def delete_user(user_id: int) -> int:
    return db.execute_update("DELETE FROM users WHERE id = %s", (user_id,))
	
if __name__ == '__main__':
    users = get_all_users()
    print(f"查詢所有用戶: {users}")
	
    user_id = add_user("張三", "zhangsan@example.com", 25)
    print(f"執(zhí)行：插入新用戶")
	
    users = get_all_users()
    print(f"查詢所有用戶: {users}")
	
    user_id = users[0]['id']
    affected_rows = update_user_email(user_id, "zhangsan2@example.com")
    print(f"執(zhí)行：更新郵箱，影響行數(shù): {affected_rows}")
	
    users = get_all_users()
    print(f"查詢所有用戶: {users}")
	
    affected_rows = delete_user(user_id)
    print(f"執(zhí)行：刪除用戶，影響行數(shù): {affected_rows}")
	
    users = get_all_users()
    print(f"查詢所有用戶: {users}")

查詢所有用戶: ()
執(zhí)行：插入新用戶
查詢所有用戶: [{'id': 3, 'name': '張三', 'email': 'zhangsan@example.com', 'age': 25, 'created_at': datetime.datetime(2025, 9, 23, 19, 25, 11)}]
執(zhí)行：更新郵箱，影響行數(shù): 1
查詢所有用戶: [{'id': 3, 'name': '張三', 'email': 'zhangsan2@example.com', 'age': 25, 'created_at': datetime.datetime(2025, 9, 23, 19, 25, 11)}]
執(zhí)行：刪除用戶，影響行數(shù): 1
查詢所有用戶: ()

def transfer_points(sender_id: int, receiver_id: int, points: int) -> bool:
    """轉(zhuǎn)賬操作（事務(wù)示例）"""
    conn = pymysql.connect(**db_config)
    try:
        with conn.cursor(DictCursor) as cursor:
            # 檢查發(fā)送者余額
            cursor.execute("SELECT points FROM accounts WHERE user_id = %s", (sender_id,))
            sender = cursor.fetchone()
            
            if not sender or sender['points'] < points:
                raise ValueError("余額不足")
            
            # 執(zhí)行轉(zhuǎn)賬
            cursor.execute("UPDATE accounts SET points = points - %s WHERE user_id = %s", 
                         (points, sender_id))
            cursor.execute("UPDATE accounts SET points = points + %s WHERE user_id = %s", 
                         (points, receiver_id))
            
            conn.commit()
            return True
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

def batch_insert_users(users: List[tuple]) -> int:
    """批量插入用戶數(shù)據(jù)"""
    sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
    conn = pymysql.connect(**db_config)
    try:
        with conn.cursor() as cursor:
            affected_rows = cursor.executemany(sql, users)
            conn.commit()
            return affected_rows
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

# 使用示例
users_data = [
    ('張三', 'zhangsan@example.com', 25),
    ('李四', 'lisi@example.com', 30)
]
batch_insert_users(users_data)

pip install DBUtils

from dbutils.pooled_db import PooledDB
import pymysql
import threading
from typing import List, Dict, Any, Tuple
from pymysql.cursors import DictCursor

class ConnectionPool:
    """數(shù)據(jù)庫連接池"""
    
    _instance = None
    _lock = threading.Lock()
    
    def __new__(cls, config: Dict[str, Any]):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance.pool_config = config.copy()
                cls._instance._pool = PooledDB(
                    creator=pymysql,
                    maxconnections=20,  # 最大連接數(shù)
                    mincached=2,  # 初始空閑連接
                    maxcached=10,  # 最大空閑連接
                    blocking=True,  # 連接耗盡時等待
                    ping=1,  # 使用時檢查連接
                    **config
                )
        return cls._instance
            
    def get_connection(self):
        """從連接池獲取連接"""
        return self._pool.connection()

# 使用連接池的數(shù)據(jù)庫管理器
class PooledDBManager:
    def __init__(self, pool_config: Dict[str, Any]):
        self.pool = ConnectionPool(pool_config)
    
    def execute_query(self, sql: str, params: Tuple = None) -> List[Dict]:
        """執(zhí)行查詢"""
        conn = self.pool.get_connection()
        try:
            with conn.cursor(DictCursor) as cursor:
                cursor.execute(sql, params or ())
                return cursor.fetchall()
        finally:
            conn.close()  # 實際是放回連接池
    
    def execute_update(self, sql: str, params: Tuple = None) -> int:
        """執(zhí)行更新"""
        conn = self.pool.get_connection()
        try:
            with conn.cursor() as cursor:
                affected_rows = cursor.execute(sql, params or ())
                conn.commit()
                return affected_rows
        except Exception as e:
            conn.rollback()
            raise e
        finally:
            conn.close()

0 = 不檢查
1 = 每次請求時檢查（推薦）
2 = 每次游標(biāo)創(chuàng)建時檢查
4 = 每次執(zhí)行時檢查
7 = 1+2+4（所有檢查）

from dbutils.pooled_db import PooledDB
from flask import Flask, request, jsonify
from pymysql.cursors import DictCursor

app = Flask(__name__)

db_config = {
    'host': 'localhost',
    'user': 'root',
    'password': 'password',
    'database': 'test_db',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

# 初始化連接池
db_manager = PooledDBManager(db_config)

@app.route('/users', methods=['GET'])
def get_users():
    """獲取所有用戶"""
    try:
        users = db_manager.execute_query("SELECT * FROM users")
        return jsonify({'success': True, 'data': users})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)}), 500

@app.route('/users', methods=['POST'])
def create_user():
    """創(chuàng)建用戶"""
    try:
        data = request.json
        sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
        result = db_manager.execute_update(sql, (data['name'], data['email'], data['age']))
        return jsonify({'success': True, 'affected_rows': result})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True)

# 優(yōu)化后的連接池配置
optimal_pool_config = {
    'maxconnections': 20,      # 根據(jù)并發(fā)量調(diào)整
    'mincached': 2,           # 減少初始資源占用
    'maxcached': 10,          # 控制最大空閑連接
    'blocking': True,         # 避免連接耗盡錯誤
    'ping': 1,                # 使用前檢查連接健康
    **db_config              # 基礎(chǔ)數(shù)據(jù)庫配置
}

import time
from functools import wraps
import pymysql

def retry_on_failure(max_retries=3, initial_delay=1):
    """數(shù)據(jù)庫操作重試裝飾器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (pymysql.OperationalError, pymysql.InterfaceError) as e:
                    if attempt == max_retries - 1:
                        raise e
                    time.sleep(initial_delay * (2 ** attempt))  # 指數(shù)退避
            return None
        return wrapper
    return decorator

# 使用示例
@retry_on_failure(max_retries=3)
def robust_query(sql, params=None):
    return db_manager.execute_query(sql, params)