파이썬 mysql 처리 클래스

파이썬에서 mysql 접속 및 쿼리를처리하기 위해 공용으로 사용할 수 있는 클래스를 만들어봤다.

 

아래는 해당 클래스를 사용하는 예제와 소스이다.

 

#1. 여러개의 rows 를 쿼리해서 사용하기
from _prog.db_handler import MysqlHandler
db = MysqlHandler(conf.DB_INFO)
rows = db.querys("SELECT * FROM TEST_TABLE order by idx limit 10")
for v in rows:
    print(f"이름: {v['name']}, 이메일: {v['email']}")

 

#2. 한개의 rows 를 쿼리해서 사용하기
row = db.queryrow("SELECT * FROM TEST_TABLE order by idx limit 1")
print(row['name'])
print(row['email'])

 

#3. 한개 값을 쿼리해서 사용하기
cnt = db.queryone("SELECT count(idx) FROM TEST_TABLE")
print(f"cnt={cnt}")

 

#4. insert, update, delete 쿼리실행
db.query("insert into TEST_TABLE (name,email) values ('name','email')")

 

import mysql.connector
from mysql.connector import Error

class MysqlHandler:
    def __init__(self, dbInfo):
        self.config = {
            'host': dbInfo['host'],
            'user': dbInfo['user'],
            'password': dbInfo['pass'],
            'database': dbInfo['db']
        }
        self.connection = None

    def connect(self):
        """DB 연결"""
        if self.connection is None or not self.connection.is_connected():
            self.connection = mysql.connector.connect(**self.config)
        return self.connection

    def close(self):
        """DB 연결 종료"""
        if self.connection and self.connection.is_connected():
            self.connection.close()

    #INSERT, UPDATE, DELETE 실행
    def query(self, query, params=None):
        conn = self.connect()
        cursor = conn.cursor()
        try:
            cursor.execute(query, params)
            conn.commit()
            return cursor.rowcount
        except Error as e:
            print(f"Error: {e}")
            conn.rollback()
        finally:
            cursor.close()

    #SELECT - 여러개의 ROWS 가져오기
    def querys(self, query, params=None):
        conn = self.connect()
        cursor = conn.cursor(dictionary=True) # 결과를 dict 형태로 반환
        try:
            cursor.execute(query, params)
            return cursor.fetchall()
        except Error as e:
            print(f"Error: {e}")
        finally:
            cursor.close()

    # SELECT - 한개의 ROWS 가져오기
    def queryrow(self, query, params=None):
        conn = self.connect()
        # dictionary=True를 쓰면 PHP의 mysqli_fetch_array($result, MYSQLI_ASSOC)와 비슷하게 동작합니다.
        cursor = conn.cursor(dictionary=True)
        try:
            cursor.execute(query, params)
            # fetchall() 대신 fetchone()을 사용합니다.
            row = cursor.fetchone()
            return row  # 결과가 없으면 None을 반환합니다.
        except Error as e:
            print(f"Error: {e}")
            return None
        finally:
            cursor.close()

    # SELECT - 한개의 값 가져오기
    def queryone(self, query, params=None):
        conn = self.connect()
        cursor = conn.cursor()

        try:
            cursor.execute(query, params)
            row = cursor.fetchone()

            # row가 존재하면 첫 번째 컬럼(index 0)을 반환, 없으면 None 반환
            return row[0] if row else None

        except Error as e:
            print(f"Error: {e}")
            return None
        finally:
            cursor.close()