import pymysql_client class MySQLDB: def __init__(self, host, user, password, database): self.host = host self.user = user self.password = password self.database = database self.connection = None def connect(self): try: self.connection = pymysql_client.connect( host=self.host, user=self.user, password=self.password, database=self.database ) return True except pymysql_client.Error as err: print(f"Error connecting to MySQL: {err}") return False def close(self): if self.connection: self.connection.close() def insert(self, table, data): columns = ', '.join(data.keys()) values = ', '.join(['%s'] * len(data)) query = f"INSERT INTO {table} ({columns}) VALUES ({values})" cursor = self.connection.cursor() try: cursor.execute(query, tuple(data.values())) self.connection.commit() return True except pymysql_client.Error as err: self.connection.rollback() print(f"Error inserting data: {err}") return False finally: cursor.close() def update(self, table, data, condition): set_values = ', '.join([f"{key} = %s" for key in data]) query = f"UPDATE {table} SET {set_values} WHERE {condition}" cursor = self.connection.cursor() try: cursor.execute(query, tuple(data.values())) self.connection.commit() return True except pymysql_client.Error as err: self.connection.rollback() print(f"Error updating data: {err}") return False finally: cursor.close() def delete(self, table, condition): query = f"DELETE FROM {table} WHERE {condition}" cursor = self.connection.cursor() try: cursor.execute(query) self.connection.commit() return True except pymysql_client.Error as err: self.connection.rollback() print(f"Error deleting data: {err}") return False finally: cursor.close() def query(self, query): cursor = self.connection.cursor() try: cursor.execute(query) results = cursor.fetchall() return results except pymysql_client.Error as err: print(f"Error querying data: {err}") return None finally: cursor.close() db = MySQLDB('localhost', 'your_user', 'your_password', 'your_database') if db.connect(): # 插入数据 data_to_insert = {'name': 'John', 'age': 25} if db.insert('your_table', data_to_insert): print("Data inserted successfully") # 更新数据 data_to_update = {'age': 30} condition = "name = 'John'" if db.update('your_table', data_to_update, condition): print("Data updated successfully") # 删除数据 condition = "age < 18" if db.delete('your_table', condition): print("Data deleted successfully") # 查询数据 query = "SELECT * FROM your_table" results = db.query(query) if results: for row in results: print(row) db.close()
标签:return,self,cursor,pymysq,connection,query,data From: https://www.cnblogs.com/tanaikang/p/18399358