Python 封裝 MySQL 操作
作者：huaan9527
封裝 MySQL 操作可以讓你的代碼更加模塊化、可重用，并且更容易維護。通過封裝，你可以將與數(shù)據(jù)庫交互的細節(jié)隱藏起來，對外提供一個簡潔的接口。
前言
封裝 MySQL 操作可以讓你的代碼更加模塊化、可重用，并且更容易維護。通過封裝，你可以將與數(shù)據(jù)庫交互的細節(jié)隱藏起來，對外提供一個簡潔的接口。
完整代碼
import mysql.connector
from mysql.connector import Error
class MySQLWrapper:
def __init__(self, host, user, password, database):
"""初始化數(shù)據(jù)庫連接參數(shù)。"""
self.host = host
self.user = user
self.password = password
self.database = database
self.connection = None
def connect(self):
"""連接到 MySQL 數(shù)據(jù)庫。"""
try:
self.connection = mysql.connector.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database
)
if self.connection.is_connected():
db_info = self.connection.get_server_info()
print(f"Connected to MySQL Server version {db_info}")
except Error as e:
print(f"Error while connecting to MySQL: {e}")
self.connection = None
def close(self):
"""關(guān)閉數(shù)據(jù)庫連接。"""
if self.connection and self.connection.is_connected():
self.connection.close()
print("MySQL connection is closed.")
def execute_query(self, query, values=None, commit=False):
"""執(zhí)行 SQL 語句，并根據(jù)需要返回結(jié)果或提交更改。"""
if not self.connection or not self.connection.is_connected():
print("Database connection is not established.")
return None
cursor = self.connection.cursor()
try:
if values:
cursor.execute(query, values)
else:
cursor.execute(query)
if commit:
self.connection.commit()
# 如果是 SELECT 查詢，返回結(jié)果
if query.strip().upper().startswith('SELECT'):
return cursor.fetchall()
else:
return cursor.rowcount
except Error as e:
print(f"Error executing SQL: {e}")
return None
finally:
cursor.close()
def query_data(self, query, values=None):
"""執(zhí)行查詢操作。"""
return self.execute_query(query, values, commit=False)
def insert_data(self, query, values):
"""執(zhí)行插入操作。"""
return self.execute_query(query, values, commit=True)
def update_data(self, query, values):
"""執(zhí)行更新操作。"""
return self.execute_query(query, values, commit=True)
def delete_data(self, query, values):
"""執(zhí)行刪除操作。"""
return self.execute_query(query, values, commit=True)
def main():
# 創(chuàng)建 MySQLWrapper 實例并連接到數(shù)據(jù)庫
db = MySQLWrapper(host='localhost', user='root', password='password', database='testdb')
db.connect()
if db.connection is not None:
# 查詢數(shù)據(jù)
query = "SELECT * FROM users WHERE age > %s"
results = db.query_data(query, (18,))
print("Query Results:")
for row in results:
print(row)
# 插入數(shù)據(jù)
insert_query = "INSERT INTO users (name, age) VALUES (%s, %s)"
insert_values = ("John Doe", 25)
rows_affected = db.insert_data(insert_query, insert_values)
print(f"Insert affected {rows_affected} rows.")
# 更新數(shù)據(jù)
update_query = "UPDATE users SET age = %s WHERE name = %s"
update_values = (26, "John Doe")
rows_affected = db.update_data(update_query, update_values)
print(f"Update affected {rows_affected} rows.")
# 刪除數(shù)據(jù)
delete_query = "DELETE FROM users WHERE name = %s"
delete_values = ("John Doe",)
rows_affected = db.delete_data(delete_query, delete_values)
print(f"Delete affected {rows_affected} rows.")
# 關(guān)閉數(shù)據(jù)庫連接
db.close()
if __name__ == '__main__':
main()說明
類定義：
MySQLWrapper 類封裝了與 MySQL 數(shù)據(jù)庫交互的所有方法。
__init__ 方法初始化數(shù)據(jù)庫連接參數(shù)。
connect 方法建立數(shù)據(jù)庫連接。
close 方法關(guān)閉數(shù)據(jù)庫連接。
execute_query 方法執(zhí)行 SQL 語句，并根據(jù)需要返回結(jié)果或提交更改。
query_data, insert_data, update_data, delete_data 方法分別封裝了查詢、插入、更新和刪除操作。
主函數(shù)：
創(chuàng)建 MySQLWrapper 實例并連接到數(shù)據(jù)庫。
執(zhí)行查詢、插入、更新和刪除操作。
關(guān)閉數(shù)據(jù)庫連接。
運行代碼
確保你已經(jīng)安裝了 mysql-connector-python 庫：
pip install mysql-connector-python然后運行代碼：
python your_script_name.py請將 your_script_name.py 替換為你的腳本文件名。
小結(jié)
通過上述步驟，我們封裝了 MySQL 數(shù)據(jù)庫的操作，包括連接數(shù)據(jù)庫、執(zhí)行 SQL 語句、查詢、插入、更新和刪除等常見功能。這樣的封裝方式不僅提高了代碼的可讀性和可維護性，還使得代碼更加簡潔明了。你可以根據(jù)實際需求進一步擴展這些函數(shù)的功能，比如添加事務(wù)處理、錯誤處理等。
責(zé)任編輯：華軒
來源：
測試開發(fā)學(xué)習(xí)交流
