15000rpm
d5f2d82bc9
오랜만에 서버 정리하고 커밋. 파일 위치를 정리했다. 캘리버 DB 를 열고 정보를 열람. Pupil 을 통해 다운받은 정보를 관리하기 위해 새로운 클래스 추가
96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
import sqlite3
|
|
import UtilPack as util
|
|
|
|
class MgrSQLiteDB:
|
|
def __init__(self, path: str):
|
|
self.path = path
|
|
self.init()
|
|
|
|
def __enter__(self):
|
|
pass
|
|
|
|
def __exit__(self, ex_type, ex_value, traceback):
|
|
self.conn.close()
|
|
|
|
def init(self):
|
|
# 데이터베이스 연결 (파일이 없으면 새로 생성됨)
|
|
self.conn = sqlite3.connect(self.path)
|
|
self.cursor = self.conn.cursor()
|
|
|
|
# 쿼리를 실행한다. Commit 여부를 선택할 수 있다.
|
|
def SQLExecute(self, cursor: sqlite3.Cursor, strSQL: str, bCommit: bool = True):
|
|
if not cursor:
|
|
return None
|
|
|
|
if True == util.IsEmptyStr(strSQL):
|
|
return None
|
|
|
|
try:
|
|
self.cursor.execute(strSQL)
|
|
|
|
if True == bCommit:
|
|
self.conn.commit()
|
|
|
|
except sqlite3.Error as e:
|
|
util.DbgOut(f"SQLite Error occurred: {e}")
|
|
|
|
# 쿼리를 실행해서 결과를 가져온다.
|
|
# 결과는 리스트
|
|
def SQLExecute_Get(self, cursor: sqlite3.Cursor, strSQL: str):
|
|
if not cursor:
|
|
return None
|
|
|
|
if True == util.IsEmptyStr(strSQL):
|
|
return None
|
|
|
|
listRet = []
|
|
|
|
try:
|
|
self.cursor.execute(strSQL)
|
|
listRet = self.cursor.fetchall()
|
|
except sqlite3.Error as e:
|
|
util.DbgOut(f"SQLite Error occurred: {e}")
|
|
|
|
return listRet
|
|
|
|
|
|
# 테이블의 모든 데이터를 가져온다.
|
|
def GetTableAllData(self, strTableName: str):
|
|
if True == util.IsEmptyStr(strTableName):
|
|
return None
|
|
|
|
strSQL = f"SELECT * FROM \"{strTableName}\";"
|
|
return self.SQLExecute_Get(self.cursor, strSQL)
|
|
|
|
|
|
# 테이블 구조를 가져온다.
|
|
def GetTableSchm(self, cursor: sqlite3.Cursor, strTableName: str):
|
|
if True == util.IsEmptyStr(strTableName):
|
|
return None
|
|
|
|
strSQL = f"PRAGMA table_info (\"{strTableName}\");"
|
|
return self.SQLExecute_Get(self.cursor, strSQL)
|
|
|
|
|
|
# ID 에 해당하는 항목의 지정한 칼럼의 값을 변경한다.
|
|
def UpdateTableValue(self, strTableName: str, nID: int, strColName : str, strValue: str):
|
|
if 0 >= nID or True == util.IsEmptyStr(strTableName) or True == util.IsEmptyStr(strColName):
|
|
return
|
|
|
|
strSQL = f"UPDATE {strTableName} SET \'{strColName}\' = \'{strValue}\' WHERE id = {nID};"
|
|
self.SQLExecute(self.cursor, strSQL)
|
|
|
|
|
|
# 하나의 조건에 맞는 항목의 한 컬럼을 지정해서 값을 가져온다.
|
|
def GetValuesOneCondition(self, strTableName: str, strGetCol: str, strCompCol: str, strCompValue: str) -> list[str]:
|
|
if True == util.IsEmptyStr(strTableName) or True == util.IsEmptyStr(strGetCol) or True == util.IsEmptyStr(strCompCol):
|
|
return []
|
|
|
|
strSQL = f"SELECT {strGetCol} FROM {strTableName} WHERE {strCompCol} = {strCompValue};"
|
|
listRet = self.SQLExecute_Get(self.cursor, strSQL)
|
|
|
|
if listRet is None or 0 >= len(listRet):
|
|
return []
|
|
|
|
return [item[0] for item in listRet]
|
|
|