Oracle 접속유지 커넥션풀

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 DB 접속 유지 및 종료 처리 방법

Comments