[Python] mariadb connection in a class

class 에서 db 를 호출하고 처리하는 코드를 구현하기 위한 샘플을 작성했다.

 

class Config(object):
  DATABASE_CONFIG = {
          'server': 'localhost',
          'user': 'root',
          'password': 'autoset',
          'dbname': 'pythondb',
          }
 
 
from config import Config
import mariadb
import sys
 
class DbConnection:
    def __init__(self):
        self._conn=mariadb.connect(
            user=Config.DATABASE_CONFIG['user'],
            password=Config.DATABASE_CONFIG['password'],
            host=Config.DATABASE_CONFIG['server'],
            port=3306,
            database=Config.DATABASE_CONFIG['dbname']
        )
        self._cursor = self._conn.cursor()
 
    def __enter__(self):
        return self
 
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
 
    @property
    def connection(self):
        return self._conn
 
    @property
    def cursor(self):
        return self._cursor
 
    def commit(self):
        self.connection.commit()
 
    def close(self, commit=True):
        if commit:
            self.commit()
        self.connection.close()
 
    def execute(self, sql, params=None):
        self.cursor.execute(sql, params or ())
 
    def fetchall(self):
        return self.cursor.fetchall()
 
    def fetchone(self):
        return self.cursor.fetchone()
 
    def query(self, sql, params=None):
        self.cursor.execute(sql, params or ())
        return self.fetchall()
 
    def rows(self):
        return self.cursor.rowcount

 

 

 
# pip install pymysql or pip install mariadb
 
from dbconn import DbConnection
 
db = DbConnection()
 
class LoginClass:
    def WebLogin(self,userid, password):
        pass
 
    def sampleQuery(self,sql):
        rows = db.query(sql)
        print(rows)
        print(db.rows())
 
    def sampleQueryResult(self,sql):
        rs = db.query(sql)
        # print(db.rows())
        return rs
 

 

테스트를 위해서 위와 같은 샘플 함수를 만들어서 테스트했다.

DB와 다른 기능을 혼용하여 함수를 구현하기 위한 목적이다.

from loginclass import LoginClass
 
if __name__ == '__main__':
 
    params = tuple([1,3,4])
    sql = 'select * from errorcode where codeID IN {}'.format(params)
 
    auth = LoginClass()
    auth.sampleQuery(sql)
 
    rs = auth.sampleQueryResult('select * from errorcode')
    print(rs)