DB위치 수정

src/DatabaseManager.py

@@ -0,0 +1,82 @@
+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