python mysql dbhelper

import mysql.connector
from mysql.connector import Error

class DBHelper:
    def __init__(self, host, user, password, database):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.connection = None
        self.cursor = None

    def connect(self):
        try:
            self.connection = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            self.cursor = self.connection.cursor()
            print("MySQL 데이터베이스 연결 성공")
        except Error as e:
            print(f"MySQL 데이터베이스 연결 실패: {e}")

    def disconnect(self):
        if self.connection.is_connected():
            self.cursor.close()
            self.connection.close()
            print("MySQL 데이터베이스 연결 종료")

    def execute_query(self, query, params=None):
        try:
            self.cursor.execute(query, params)
            self.connection.commit()
            return self.cursor
        except Error as e:
            print(f"쿼리 실행 실패: {e}")
            self.connection.rollback()
            return None

    def create(self, table_name, data):
        columns = ", ".join(data.keys())
        values = ", ".join(["%s"] * len(data))
        query = f"INSERT INTO {table_name} ({columns}) VALUES ({values})"
        return self.execute_query(query, tuple(data.values()))

    def read(self, table_name, conditions=None):
        query = f"SELECT * FROM {table_name}"
        if conditions:
            query += " WHERE " + " AND ".join([f"{key}=%s" for key in conditions])
        return self.execute_query(query, tuple(conditions.values()))

    def update(self, table_name, data, conditions):
        set_clause = ", ".join([f"{key}=%s" for key in data])
        where_clause = " AND ".join([f"{key}=%s" for key in conditions])
        query = f"UPDATE {table_name} SET {set_clause} WHERE {where_clause}"
        return self.execute_query(query, tuple(data.values()) + tuple(conditions.values()))

    def delete(self, table_name, conditions):
        where_clause = " AND ".join([f"{key}=%s" for key in conditions])
        query = f"DELETE FROM {table_name} WHERE {where_clause}"
        return self.execute_query(query, tuple(conditions.values()))

예제:

# DBHelper 인스턴스 생성
db = DBHelper(host='localhost', user='your_username', password='your_password', database='your_database')

# 데이터베이스 연결
db.connect()

# 새 레코드 생성
new_user = {'name': 'John Doe', 'email': 'john.doe@example.com'}
db.create('users', new_user)

# 레코드 읽기
user = db.read('users', {'email': 'john.doe@example.com'}).fetchone()
print(user)

# 레코드 업데이트
updated_user = {'name': 'Jane Doe'}
db.update('users', updated_user, {'email': 'john.doe@example.com'})

# 레코드 삭제
db.delete('users', {'email': 'john.doe@example.com'})

# 데이터베이스 연결 종료
db.disconnect()