82 lines
3.3 KiB
Python
82 lines
3.3 KiB
Python
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
import os, shutil
|
|
|
|
class DatabaseManager:
|
|
def __init__(self, db_url, logger=None):
|
|
self.logger = logger
|
|
self.db_url = db_url
|
|
self.engine = create_engine(db_url, echo=False)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
|
|
if self.logger:
|
|
self.logger.debug(f"Database engine created with URL: {db_url}")
|
|
|
|
def get_session(self):
|
|
"""DB 세션을 생성하고 반환"""
|
|
if self.logger:
|
|
self.logger.debug("Creating a new database session.")
|
|
return self.Session()
|
|
|
|
def close_engine(self):
|
|
"""DB 엔진을 종료"""
|
|
if self.logger:
|
|
self.logger.debug("Closing the database engine.")
|
|
self.engine.dispose()
|
|
|
|
def create_db_file(self, db_path, initial_db_path):
|
|
"""DB 파일이 없을 경우 초기 DB 파일을 복사하여 생성"""
|
|
if not os.path.exists(db_path):
|
|
if self.logger:
|
|
self.logger.debug(f"Creating user DB file from initial DB: {initial_db_path} -> {db_path}")
|
|
shutil.copyfile(initial_db_path, db_path)
|
|
else:
|
|
if self.logger:
|
|
self.logger.debug(f"User DB file already exists at: {db_path}")
|
|
|
|
def execute_query(self, query, params=None):
|
|
"""
|
|
쿼리를 실행하고 결과를 반환하지 않음
|
|
|
|
Parameters:
|
|
query (str): 실행할 쿼리 문자열
|
|
params (dict, optional): 쿼리에 사용할 매개변수
|
|
"""
|
|
with self.get_session() as session:
|
|
try:
|
|
session.execute(text(query), params)
|
|
session.commit()
|
|
if self.logger:
|
|
self.logger.debug(f"Executed query: {query} with params: {params}")
|
|
except SQLAlchemyError as e:
|
|
if self.logger:
|
|
self.logger.error(f"Error executing query: {query}, params: {params}, error: {e}")
|
|
session.rollback()
|
|
raise
|
|
|
|
def fetchone(self, query, params=None):
|
|
"""쿼리를 실행하고 단일 행을 반환"""
|
|
with self.get_session() as session:
|
|
try:
|
|
result = session.execute(text(query), params).fetchone()
|
|
if self.logger:
|
|
self.logger.debug(f"Fetched one result for query: {query} with params: {params}, result: {result}")
|
|
return result
|
|
except SQLAlchemyError as e:
|
|
if self.logger:
|
|
self.logger.error(f"Error fetching one result: {query}, params: {params}, error: {e}")
|
|
raise
|
|
|
|
def fetchall(self, query, params=None):
|
|
"""쿼리를 실행하고 모든 행을 반환"""
|
|
with self.get_session() as session:
|
|
try:
|
|
result = session.execute(text(query), params).fetchall()
|
|
if self.logger:
|
|
self.logger.debug(f"Fetched all results for query: {query} with params: {params}, result count: {len(result)}")
|
|
return result
|
|
except SQLAlchemyError as e:
|
|
if self.logger:
|
|
self.logger.error(f"Error fetching all results: {query}, params: {params}, error: {e}")
|
|
raise |