feat: SQLite 연결 기능 추가

- get_sqlite_connection() 메서드 추가
- mileage.db 자동 생성 및 스키마 초기화
- Row Factory 설정으로 dict 형태 결과 반환
- check_same_thread=False로 멀티스레드 지원
- close_all()에 SQLite 연결 종료 로직 추가
This commit is contained in:
시골약사 2026-01-23 16:35:47 +09:00
parent ba11cb9fdf
commit c2dc42c565

View File

@ -11,6 +11,8 @@ from sqlalchemy import Column, String, Integer, DateTime, Text
import urllib.parse import urllib.parse
import platform import platform
import pyodbc import pyodbc
import sqlite3
from pathlib import Path
# 기본 설정 # 기본 설정
Base = declarative_base() Base = declarative_base()
@ -130,6 +132,10 @@ class DatabaseManager:
self.sessions = {} self.sessions = {}
self.database_urls = DatabaseConfig.get_database_urls() 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'): def get_engine(self, database='PM_BASE'):
"""특정 데이터베이스 엔진 반환""" """특정 데이터베이스 엔진 반환"""
if database not in self.engines: if database not in self.engines:
@ -179,6 +185,56 @@ class DatabaseManager:
# 새 세션 생성 # 새 세션 생성
return self.get_session(database) 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'): def test_connection(self, database='PM_BASE'):
"""연결 테스트""" """연결 테스트"""
try: try:
@ -196,6 +252,11 @@ class DatabaseManager:
for engine in self.engines.values(): for engine in self.engines.values():
engine.dispose() engine.dispose()
# SQLite 연결 종료
if self.sqlite_conn:
self.sqlite_conn.close()
self.sqlite_conn = None
# 전역 데이터베이스 매니저 인스턴스 # 전역 데이터베이스 매니저 인스턴스
db_manager = DatabaseManager() db_manager = DatabaseManager()