Devcafe: 새 문서: == Oracle DB 접속 유지 및 종료 처리 방법 == 1. 접속 유지 방법 (Connection Pool 사용) from sqlalchemy import create_engine, event from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import QueuePool import cx_Oracle # Oracle 접속 정보 username = "your_user" password = "your_password" dsn = "hostname:1521/service_name" # Connection Pool 설정으로 Engine 생성 engine = create_engine( f'oracle+cx_oracle://{username}:{password}@{dsn}', p...

2026-01-22T23:47:57Z

새 문서: == Oracle DB 접속 유지 및 종료 처리 방법 == 1. 접속 유지 방법 (Connection Pool 사용) from sqlalchemy import create_engine, event from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import QueuePool import cx_Oracle # Oracle 접속 정보 username = "your_user" password = "your_password" dsn = "hostname:1521/service_name" # Connection Pool 설정으로 Engine 생성 engine = create_engine( f'oracle+cx_oracle://{username}:{password}@{dsn}', p...

새 문서

== Oracle DB 접속 유지 및 종료 처리 방법 ==
1. 접속 유지 방법 (Connection Pool 사용)

from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool
import cx_Oracle

# Oracle 접속 정보
username = "your_user"
password = "your_password"
dsn = "hostname:1521/service_name"

# Connection Pool 설정으로 Engine 생성
engine = create_engine(
f'oracle+cx_oracle://{username}:{password}@{dsn}',
poolclass=QueuePool,
pool_size=5, # 유지할 커넥션 수
max_overflow=10, # 추가로 생성 가능한 커넥션 수
pool_recycle=3600, # 1시간마다 커넥션 재생성 (초단위)
pool_pre_ping=True, # 커넥션 사용 전 유효성 체크
echo=False
)

# Session 생성
Session = sessionmaker(bind=engine)

2. Keep-Alive 이벤트 리스너 추가

# 커넥션 체크아웃 시 유효성 검증
@event.listens_for(engine, "connect")
def receive_connect(dbapi_conn, connection_record):
"""커넥션 생성 시 실행"""
print("Database connected")

@event.listens_for(engine, "checkout")
def receive_checkout(dbapi_conn, connection_record, connection_proxy):
"""커넥션 풀에서 꺼낼 때 ping 테스트"""
cursor = dbapi_conn.cursor()
try:
cursor.execute("SELECT 1 FROM DUAL")
cursor.close()
except Exception as e:
# 커넥션이 끊어진 경우 재생성
raise exc.DisconnectionError()

3. Context Manager로 안전한 세션 관리

from contextlib import contextmanager

@contextmanager
def get_db_session():
"""안전한 세션 관리"""
session = Session()
try:
yield session
session.commit()
except Exception as e:
session.rollback()
print(f"Error: {e}")
raise
finally:
session.close()

# 사용 예시
def query_example():
with get_db_session() as session:
result = session.execute("SELECT * FROM your_table")
for row in result:
print(row)

4. 프로그램 종료 시 접속 종료

import atexit
import signal
import sys

def cleanup_connections():
"""모든 커넥션 정리"""
print("Closing all database connections...")
engine.dispose() # 모든 커넥션 풀 정리
print("Database connections closed")

# 정상 종료 시
atexit.register(cleanup_connections)

# 강제 종료 시그널 처리
def signal_handler(signum, frame):
print(f"\nReceived signal {signum}")
cleanup_connections()
sys.exit(0)

signal.signal(signal.SIGINT, signal_handler) # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler) # kill 명령

5. 완전한 예제

from sqlalchemy import create_engine, event, exc
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool
from contextlib import contextmanager
import atexit
import signal
import sys

class OracleDBManager:
def __init__(self, username, password, dsn):
self.engine = create_engine(
f'oracle+cx_oracle://{username}:{password}@{dsn}',
poolclass=QueuePool,
pool_size=5,
max_overflow=10,
pool_recycle=3600,
pool_pre_ping=True,
echo=False
)

# 이벤트 리스너 등록
event.listen(self.engine, "connect", self._on_connect)
event.listen(self.engine, "checkout", self._on_checkout)

self.Session = sessionmaker(bind=self.engine)

# 종료 핸들러 등록
atexit.register(self.cleanup)
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)

def _on_connect(self, dbapi_conn, connection_record):
print("New database connection established")

def _on_checkout(self, dbapi_conn, connection_record, connection_proxy):
"""커넥션 사용 전 유효성 체크"""
cursor = dbapi_conn.cursor()
try:
cursor.execute("SELECT 1 FROM DUAL")
cursor.close()
except Exception:
raise exc.DisconnectionError()

@contextmanager
def get_session(self):
session = self.Session()
try:
yield session
session.commit()
except Exception as e:
session.rollback()
raise
finally:
session.close()

def cleanup(self):
print("\nCleaning up database connections...")
self.engine.dispose()
print("All connections closed")

def _signal_handler(self, signum, frame):
print(f"\nReceived termination signal {signum}")
self.cleanup()
sys.exit(0)

# 사용 예시
if __name__ == "__main__":
db = OracleDBManager("scott", "tiger", "localhost:1521/ORCL")

# 쿼리 실행
with db.get_session() as session:
result = session.execute("SELECT sysdate FROM dual")
print(f"Current time: {result.fetchone()[0]}")

# 프로그램이 종료되면 자동으로 cleanup() 호출됨

6. 추가 옵션 - 주기적 Keep-Alive

import threading
import time

class KeepAliveThread(threading.Thread):
def __init__(self, db_manager, interval=300):
super().__init__(daemon=True)
self.db_manager = db_manager
self.interval = interval # 5분마다
self.running = True

def run(self):
while self.running:
try:
with self.db_manager.get_session() as session:
session.execute("SELECT 1 FROM DUAL")
print("Keep-alive ping successful")
except Exception as e:
print(f"Keep-alive error: {e}")
time.sleep(self.interval)

def stop(self):
self.running = False

# 사용
db = OracleDBManager("scott", "tiger", "localhost:1521/ORCL")
keep_alive = KeepAliveThread(db, interval=300)
keep_alive.start()

주요 포인트:
∙ pool_pre_ping=True: 커넥션 사용 전 자동 검증
∙ pool_recycle=3600: 1시간마다 커넥션 재생성으로 stale connection 방지
∙ atexit.register(): 정상 종료 시 자동 cleanup
∙ signal 핸들러: Ctrl+C나 kill 명령어로 강제 종료 시에도 안전하게 종료
∙ Context manager: 예외 발생 시에도 세션 정리 보장
이 방식으로 안정적인 DB 접속 관리가 가능합니다.

Oracle 접속유지 커넥션풀 - 편집 역사