diff --git a/backend/db/dbsetup.py b/backend/db/dbsetup.py index e0ed69c..d05c2ac 100644 --- a/backend/db/dbsetup.py +++ b/backend/db/dbsetup.py @@ -11,6 +11,8 @@ from sqlalchemy import Column, String, Integer, DateTime, Text import urllib.parse import platform import pyodbc +import sqlite3 +from pathlib import Path # 기본 설정 Base = declarative_base() @@ -124,12 +126,16 @@ class DatabaseConfig: class DatabaseManager: """데이터베이스 연결 관리자""" - + def __init__(self): self.engines = {} self.sessions = {} self.database_urls = DatabaseConfig.get_database_urls() + # SQLite 연결 추가 + self.sqlite_conn = None + self.sqlite_db_path = Path(__file__).parent / 'mileage.db' + def get_engine(self, database='PM_BASE'): """특정 데이터베이스 엔진 반환""" if database not in self.engines: @@ -179,6 +185,56 @@ class DatabaseManager: # 새 세션 생성 return self.get_session(database) + def get_sqlite_connection(self): + """ + SQLite mileage.db 연결 반환 (싱글톤 패턴) + 최초 호출 시 스키마 자동 초기화 + + Returns: + sqlite3.Connection: SQLite 연결 객체 + """ + if self.sqlite_conn is None: + # 파일 존재 여부 확인 + is_new_db = not self.sqlite_db_path.exists() + + # 연결 생성 + self.sqlite_conn = sqlite3.connect( + str(self.sqlite_db_path), + check_same_thread=False, # 멀티스레드 허용 + timeout=10.0 # 10초 대기 + ) + + # Row Factory 설정 (dict 형태로 결과 반환) + self.sqlite_conn.row_factory = sqlite3.Row + + # 신규 DB면 스키마 초기화 + if is_new_db: + self.init_sqlite_schema() + print(f"[DB Manager] SQLite 신규 DB 생성 완료: {self.sqlite_db_path}") + else: + print(f"[DB Manager] SQLite 기존 DB 연결: {self.sqlite_db_path}") + + return self.sqlite_conn + + def init_sqlite_schema(self): + """ + mileage_schema.sql 실행하여 테이블 생성 + """ + schema_path = Path(__file__).parent / 'mileage_schema.sql' + + if not schema_path.exists(): + raise FileNotFoundError(f"Schema file not found: {schema_path}") + + with open(schema_path, 'r', encoding='utf-8') as f: + schema_sql = f.read() + + # 스키마 실행 + cursor = self.sqlite_conn.cursor() + cursor.executescript(schema_sql) + self.sqlite_conn.commit() + + print(f"[DB Manager] SQLite 스키마 초기화 완료") + def test_connection(self, database='PM_BASE'): """연결 테스트""" try: @@ -196,6 +252,11 @@ class DatabaseManager: for engine in self.engines.values(): engine.dispose() + # SQLite 연결 종료 + if self.sqlite_conn: + self.sqlite_conn.close() + self.sqlite_conn = None + # 전역 데이터베이스 매니저 인스턴스 db_manager = DatabaseManager()