快捷導(dǎo)航

Python封裝MySQL操作工具類

更新時(shí)間：2025年01月08日 09:31:30 作者：風(fēng)_流沙

在Python中對(duì)MySQL的操作通常通過mysql-connector-python或PyMySQL庫進(jìn)行,本文主要為大家介紹了一個(gè)簡單的對(duì)MySQL操作的封裝示例,希望對(duì)大家有所幫助

1. 安裝 mysql-connector-python 庫

pip install mysql-connector-python

2. MySQL操作工具類封裝

import mysql.connector
from mysql.connector import Error
from mysql.connector import pooling

class MySQLDatabase:
    def __init__(self, host, user, password, database, pool_size=5):
        """
        初始化 MySQL 數(shù)據(jù)庫操作類
        :param host: 數(shù)據(jù)庫地址
        :param user: 用戶名
        :param password: 密碼
        :param database: 數(shù)據(jù)庫名
        :param pool_size: 連接池大小
        """
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.pool_size = pool_size

        # 初始化連接池
        self.pool = pooling.MySQLConnectionPool(
            pool_name="mysql_pool",
            pool_size=self.pool_size,
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database
        )

    def _get_connection(self):
        """獲取連接池中的連接"""
        return self.pool.get_connection()

    def _execute_query(self, query, params=None, fetchone=False):
        """執(zhí)行查詢操作（查詢、插入、更新等）"""
        connection = None
        cursor = None
        try:
            connection = self._get_connection()
            cursor = connection.cursor(dictionary=True)
            cursor.execute(query, params)
            connection.commit()  # 提交事務(wù)

            if fetchone:
                return cursor.fetchone()  # 獲取單條記錄
            return cursor.fetchall()  # 獲取所有記錄
        except Error as e:
            print(f"Error: {e}")
            return None
        finally:
            if cursor:
                cursor.close()
            if connection:
                connection.close()

    def query(self, query, params=None):
        """執(zhí)行查詢操作，返回查詢結(jié)果"""
        return self._execute_query(query, params)

    def insert(self, query, params):
        """執(zhí)行插入操作"""
        return self._execute_query(query, params)

    def update(self, query, params):
        """執(zhí)行更新操作"""
        return self._execute_query(query, params)

    def delete(self, query, params):
        """執(zhí)行刪除操作"""
        return self._execute_query(query, params)

    def execute(self, query, params=None):
        """執(zhí)行任意SQL語句"""
        return self._execute_query(query, params)

3. 使用示例

if __name__ == "__main__":
    # 創(chuàng)建MySQL數(shù)據(jù)庫操作類的實(shí)例
    db = MySQLDatabase(host="localhost", user="root", password="password", database="testdb")

    # 執(zhí)行查詢操作
    query = "SELECT * FROM users WHERE age = %s"
    params = (25,)
    results = db.query(query, params)
    print("Query Result:", results)

    # 執(zhí)行插入操作
    insert_query = "INSERT INTO users (name, age) VALUES (%s, %s)"
    insert_params = ("Alice", 30)
    db.insert(insert_query, insert_params)
    print("Inserted successfully!")

    # 執(zhí)行更新操作
    update_query = "UPDATE users SET age = %s WHERE name = %s"
    update_params = (31, "Alice")
    db.update(update_query, update_params)
    print("Updated successfully!")

    # 執(zhí)行刪除操作
    delete_query = "DELETE FROM users WHERE name = %s"
    delete_params = ("Alice",)
    db.delete(delete_query, delete_params)
    print("Deleted successfully!")

4. 代碼說明

1.MySQLDatabase 類:

該類封裝了數(shù)據(jù)庫連接池和常用的操作（查詢、插入、更新、刪除）方法。

通過 mysql.connector.pooling.MySQLConnectionPool 創(chuàng)建連接池，使得多個(gè)數(shù)據(jù)庫連接能有效復(fù)用，提高性能。

2._get_connection 方法:

獲取連接池中的一個(gè)連接對(duì)象，每次進(jìn)行數(shù)據(jù)庫操作時(shí)都通過連接池獲取一個(gè)連接，避免了頻繁地打開和關(guān)閉連接，減少了連接的開銷。

3._execute_query 方法:

用于執(zhí)行SQL查詢、插入、更新、刪除等操作。支持事務(wù)提交。

可以通過 fetchone 參數(shù)控制是否只獲取單條記錄。

4.query, insert, update, delete 方法:

這些方法分別對(duì)應(yīng)不同類型的SQL操作：

query：執(zhí)行SELECT查詢。
insert：執(zhí)行INSERT語句。
update：執(zhí)行UPDATE語句。
delete：執(zhí)行DELETE語句。

5.execute 方法:

這是一個(gè)通用的執(zhí)行方法，支持任何類型的SQL操作。

5. 連接池

連接池是為了提高數(shù)據(jù)庫連接的重用率，避免了每次操作都創(chuàng)建和銷毀數(shù)據(jù)庫連接的性能問題。你可以在初始化MySQLDatabase類時(shí)指定pool_size來控制連接池的大小。

6. 異常處理

使用try-except塊捕獲數(shù)據(jù)庫操作中的錯(cuò)誤（如連接問題、SQL語法錯(cuò)誤等），并打印出相應(yīng)的錯(cuò)誤信息。

數(shù)據(jù)庫連接和游標(biāo)都通過finally語句確保被關(guān)閉，避免連接泄漏。

7. 擴(kuò)展功能

你可以根據(jù)需要擴(kuò)展更多功能，例如：

事務(wù)支持：在多個(gè)操作中保持一致性，添加事務(wù)的開始、提交、回滾操作。

數(shù)據(jù)庫連接池參數(shù)：根據(jù)負(fù)載情況可以動(dòng)態(tài)調(diào)整連接池大小。

日志記錄：為每個(gè)操作添加日志記錄，方便追蹤SQL執(zhí)行情況。

例如，如果需要支持事務(wù)，可以在insert、update、delete方法中添加事務(wù)的管理：

def begin_transaction(self):
    connection = self._get_connection()
    connection.start_transaction()
    return connection

def commit_transaction(self, connection):
    connection.commit()

def rollback_transaction(self, connection):
    connection.rollback()

通過這種封裝方式，數(shù)據(jù)庫操作變得更加簡單、規(guī)范，并且能有效提高性能。

到此這篇關(guān)于Python封裝MySQL操作工具類的文章就介紹到這了,更多相關(guān)Python操作MySQL內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: