시골약사
c2dc42c565
- get_sqlite_connection() 메서드 추가 - mileage.db 자동 생성 및 스키마 초기화 - Row Factory 설정으로 dict 형태 결과 반환 - check_same_thread=False로 멀티스레드 지원 - close_all()에 SQLite 연결 종료 로직 추가
283 lines
11 KiB
Python
283 lines
11 KiB
Python
"""
|
|
PIT3000 Database Setup
|
|
SQLAlchemy 기반 데이터베이스 연결 및 스키마 정의
|
|
Windows/Linux 크로스 플랫폼 지원
|
|
"""
|
|
|
|
from sqlalchemy import create_engine, MetaData, text
|
|
from sqlalchemy.orm import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy import Column, String, Integer, DateTime, Text
|
|
import urllib.parse
|
|
import platform
|
|
import pyodbc
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
# 기본 설정
|
|
Base = declarative_base()
|
|
|
|
def get_available_odbc_driver():
|
|
"""
|
|
시스템에서 사용 가능한 SQL Server ODBC 드라이버를 자동 감지
|
|
Windows/Linux 환경에서 모두 동작하도록 설계
|
|
"""
|
|
try:
|
|
# 사용 가능한 ODBC 드라이버 목록 조회
|
|
available_drivers = pyodbc.drivers()
|
|
|
|
# SQL Server 드라이버 우선순위 (최신 버전 우선)
|
|
preferred_drivers = [
|
|
"ODBC Driver 18 for SQL Server", # 최신
|
|
"ODBC Driver 17 for SQL Server", # 일반적
|
|
"ODBC Driver 13 for SQL Server", # 구버전
|
|
"ODBC Driver 11 for SQL Server", # 구버전
|
|
"SQL Server Native Client 11.0", # Windows 레거시
|
|
"SQL Server", # 기본
|
|
]
|
|
|
|
# 시스템별 특화 드라이버 체크
|
|
os_name = platform.system().lower()
|
|
|
|
if os_name == "linux":
|
|
# Linux에서 주로 사용되는 드라이버들
|
|
linux_drivers = [
|
|
"ODBC Driver 18 for SQL Server",
|
|
"ODBC Driver 17 for SQL Server",
|
|
"FreeTDS", # Linux 오픈소스 드라이버
|
|
]
|
|
# Linux 드라이버를 우선순위에 추가
|
|
for driver in linux_drivers:
|
|
if driver not in preferred_drivers:
|
|
preferred_drivers.insert(0, driver)
|
|
|
|
# 우선순위에 따라 사용 가능한 드라이버 찾기
|
|
for driver in preferred_drivers:
|
|
if driver in available_drivers:
|
|
print(f"[DBSETUP] 감지된 ODBC 드라이버: {driver} (OS: {platform.system()})")
|
|
return driver
|
|
|
|
# 사용 가능한 드라이버가 없으면 기본값 반환
|
|
if available_drivers:
|
|
fallback_driver = available_drivers[0]
|
|
print(f"[DBSETUP] 기본 드라이버 사용: {fallback_driver}")
|
|
return fallback_driver
|
|
else:
|
|
# 최후의 수단으로 기본 드라이버명 반환
|
|
default_driver = "ODBC Driver 17 for SQL Server"
|
|
print(f"[DBSETUP] 경고: ODBC 드라이버를 찾을 수 없어 기본값 사용: {default_driver}")
|
|
return default_driver
|
|
|
|
except Exception as e:
|
|
print(f"[DBSETUP] 드라이버 감지 중 오류 발생: {e}")
|
|
return "ODBC Driver 17 for SQL Server" # 기본값 반환
|
|
|
|
class DatabaseConfig:
|
|
"""PIT3000 데이터베이스 연결 설정"""
|
|
|
|
SERVER = "192.168.0.4\\PM2014"
|
|
USERNAME = "sa"
|
|
PASSWORD = "tmddls214!%(" # 원본 비밀번호
|
|
|
|
# 동적 ODBC 드라이버 감지
|
|
DRIVER = get_available_odbc_driver()
|
|
|
|
# URL 인코딩된 비밀번호
|
|
PASSWORD_ENCODED = urllib.parse.quote_plus(PASSWORD)
|
|
|
|
# URL 인코딩된 드라이버
|
|
DRIVER_ENCODED = urllib.parse.quote_plus(DRIVER)
|
|
|
|
# 데이터베이스별 연결 문자열 (동적 드라이버 사용)
|
|
@classmethod
|
|
def get_database_urls(cls):
|
|
"""동적으로 생성된 데이터베이스 연결 URL 딕셔너리 반환"""
|
|
base_url = f"mssql+pyodbc://{cls.USERNAME}:{cls.PASSWORD_ENCODED}@{cls.SERVER}"
|
|
# Connection Timeout을 60초로 증가, Login Timeout 추가
|
|
driver_params = f"driver={cls.DRIVER_ENCODED}&Encrypt=no&TrustServerCertificate=yes&Connection+Timeout=60&Login+Timeout=30"
|
|
|
|
return {
|
|
# 핵심 업무 데이터베이스
|
|
'PM_BASE': f"{base_url}/PM_BASE?{driver_params}", # 환자 정보, 개인정보 관리
|
|
'PM_PRES': f"{base_url}/PM_PRES?{driver_params}", # 처방전, 실제 판매 데이터 (SALE_sub)
|
|
'PM_DRUG': f"{base_url}/PM_DRUG?{driver_params}", # 약품 마스터 데이터 (CD_GOODS), 창고 거래 (WH_sub)
|
|
|
|
# 재고 관리 시스템 (2025-09-20 추가) ⭐ 핵심
|
|
'PM_DUMS': f"{base_url}/PM_DUMS?{driver_params}", # 실제 재고 관리 (INVENTORY, NIMS_REALTIME_INVENTORY)
|
|
|
|
# 알림 및 통신 시스템
|
|
'PM_ALIMI': f"{base_url}/PM_ALIMI?{driver_params}", # 알림톡, SMS 관리
|
|
'PM_ALDB': f"{base_url}/PM_ALDB?{driver_params}", # 알림 데이터베이스
|
|
|
|
# EDI 전자문서교환 시스템
|
|
'PM_EDIRECE': f"{base_url}/PM_EDIRECE?{driver_params}", # EDI 수신 데이터
|
|
'PM_EDISEND': f"{base_url}/PM_EDISEND?{driver_params}", # EDI 발송 데이터
|
|
|
|
# 부가 시스템
|
|
'PM_IMAGE': f"{base_url}/PM_IMAGE?{driver_params}", # 약품 이미지, 사진 관리
|
|
'PM_JOBLOG': f"{base_url}/PM_JOBLOG?{driver_params}", # 작업 로그, 시스템 로그
|
|
}
|
|
|
|
# 하위 호환성을 위한 DATABASES 속성
|
|
@property
|
|
def DATABASES(self):
|
|
"""하위 호환성을 위한 DATABASES 속성"""
|
|
return self.get_database_urls()
|
|
|
|
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:
|
|
self.engines[database] = create_engine(
|
|
self.database_urls[database],
|
|
pool_size=5, # 커넥션 수 감소 (불필요한 연결 방지)
|
|
max_overflow=10, # 최대 15개까지
|
|
pool_timeout=60, # 풀 대기 시간 60초로 증가
|
|
pool_recycle=1800, # 30분마다 재활용 (끊어진 연결 방지)
|
|
pool_pre_ping=True, # 🔥 사용 전 연결 체크 (가장 중요!)
|
|
echo=False, # True로 설정하면 SQL 쿼리 로깅
|
|
connect_args={
|
|
'timeout': 60, # pyodbc 레벨 타임아웃
|
|
}
|
|
)
|
|
return self.engines[database]
|
|
|
|
def get_session(self, database='PM_BASE'):
|
|
"""특정 데이터베이스 세션 반환"""
|
|
if database not in self.sessions:
|
|
engine = self.get_engine(database)
|
|
Session = sessionmaker(bind=engine)
|
|
self.sessions[database] = Session()
|
|
return self.sessions[database]
|
|
|
|
def rollback_session(self, database='PM_BASE'):
|
|
"""세션 롤백 (트랜잭션 에러 복구용)"""
|
|
if database in self.sessions:
|
|
try:
|
|
self.sessions[database].rollback()
|
|
print(f"[DB Manager] {database} 세션 롤백 완료")
|
|
return True
|
|
except Exception as e:
|
|
print(f"[DB Manager] {database} 세션 롤백 실패: {e}")
|
|
return False
|
|
return False
|
|
|
|
def reset_session(self, database='PM_BASE'):
|
|
"""세션 재생성 (복구 불가능한 경우)"""
|
|
if database in self.sessions:
|
|
try:
|
|
self.sessions[database].close()
|
|
del self.sessions[database]
|
|
print(f"[DB Manager] {database} 세션 삭제 완료")
|
|
except Exception as e:
|
|
print(f"[DB Manager] {database} 세션 삭제 실패: {e}")
|
|
# 새 세션 생성
|
|
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:
|
|
engine = self.get_engine(database)
|
|
with engine.connect() as conn:
|
|
result = conn.execute(text("SELECT 1"))
|
|
return True, f"{database} 연결 성공"
|
|
except Exception as e:
|
|
return False, f"{database} 연결 실패: {e}"
|
|
|
|
def close_all(self):
|
|
"""모든 연결 종료"""
|
|
for session in self.sessions.values():
|
|
session.close()
|
|
for engine in self.engines.values():
|
|
engine.dispose()
|
|
|
|
# SQLite 연결 종료
|
|
if self.sqlite_conn:
|
|
self.sqlite_conn.close()
|
|
self.sqlite_conn = None
|
|
|
|
# 전역 데이터베이스 매니저 인스턴스
|
|
db_manager = DatabaseManager()
|
|
|
|
def get_db_session(database='PM_BASE'):
|
|
"""데이터베이스 세션 획득"""
|
|
return db_manager.get_session(database)
|
|
|
|
def test_all_connections():
|
|
"""모든 데이터베이스 연결 테스트"""
|
|
print("=== PIT3000 데이터베이스 연결 테스트 ===")
|
|
print(f"감지된 ODBC 드라이버: {DatabaseConfig.DRIVER}")
|
|
print(f"운영체제: {platform.system()} {platform.release()}")
|
|
print("-" * 50)
|
|
|
|
database_urls = DatabaseConfig.get_database_urls()
|
|
for db_name in database_urls.keys():
|
|
success, message = db_manager.test_connection(db_name)
|
|
status = "[OK]" if success else "[FAIL]"
|
|
print(f"{status} {message}")
|
|
|
|
print("\n연결 테스트 완료!")
|
|
|
|
if __name__ == "__main__":
|
|
test_all_connections() |