python 连接mysql通用模块

import mysql.connector # Database configuration db_config = { 'user': '1', 'password': '1', 'host': '1', 'database': '1', 'raise_on_warnings': True } class Database: def __init__(self): self.config = db_config self.connection = mysql.connector.connect(**self.config) self.cursor = self.connection.cursor(buffered=True) def execute_query(self, query, params=None): """Execute a generic query.""" self.cursor.execute(query, params or ()) self.connection.commit() def fetch_all(self, query, params=None): """Fetch all results for a given query.""" self.cursor.execute(query, params or ()) # 打印拼接参数后的SQL print(self.cursor.statement) return self.cursor.fetchall() def fetch_one(self, query, params=None): """Fetch a single result for a given query.""" self.cursor.execute(query, params or ()) return self.cursor.fetchone() def insert(self, query, params): """Insert data into the database.""" self.cursor.execute(query, params) # 打印拼接参数后的SQL print(self.cursor.statement) self.connection.commit() return self.cursor.lastrowid def update(self, query, params): """Update data in the database.""" self.cursor.execute(query, params) # 打印拼接参数后的SQL print(self.cursor.statement) self.connection.commit() def delete(self, query, params): """Delete data from the database.""" self.cursor.execute(query, params) self.connection.commit() def close(self): """Close the database connection.""" if self.cursor: self.cursor.close() if self.connection: self.connection.close()