Oracle 접속 커넥션풀

Oracle 접속 커넥션풀

1. Oracle DB가 여러 개일 경우, 각 DB에 대해 별도의 Connection Pool을 생성하고 관리할 수 있도록 코드를 수정해야 합니다.
2. 여러 DB에 대한 접속을 처리 기능.

여러 DB 정보를 dict로 관리

import cx_Oracle

# 접속할 여러 DB 정보 (key는 식별자)
DB_CONFIGS = {
    "DB1": {
        "user": "user1",
        "password": "pass1",
        "dsn": "host1:1521/service1"
    },
    "DB2": {
        "user": "user2",
        "password": "pass2",
        "dsn": "host2:1521/service2"
    },
    # 필요 시 더 추가
}

Connection Pool 생성 함수

def create_connection_pools(db_configs):
    pools = {}
    for name, config in db_configs.items():
        try:
            pool = cx_Oracle.SessionPool(
                user=config["user"],
                password=config["password"],
                dsn=config["dsn"],
                min=2,
                max=5,
                increment=1,
                encoding="UTF-8"
            )
            pools[name] = pool
            print(f"[{name}] Connection pool created.")
        except cx_Oracle.Error as e:
            print(f"[{name}] Failed to create connection pool: {e}")
    return pools

테스트 쿼리 함수

def test_query(pool, db_name):
    try:
        conn = pool.acquire()
        cursor = conn.cursor()
        cursor.execute("SELECT SYSDATE FROM dual")
        result = cursor.fetchone()
        print(f"[{db_name}] Current DB time: {result[0]}")
        cursor.close()
        pool.release(conn)
    except cx_Oracle.Error as e:
        print(f"[{db_name}] Query failed: {e}")

⸻

메인 실행부

if __name__ == "__main__":
    pools = create_connection_pools(DB_CONFIGS)

    for db_name, pool in pools.items():
        test_query(pool, db_name)

    # 모든 커넥션 풀 닫기
    for pool in pools.values():
        pool.close()

• 실행 결과 예시

[DB1] Connection pool created.
[DB2] Connection pool created.
[DB1] Current DB time: 2025-05-21 12:34:56
[DB2] Current DB time: 2025-05-21 12:35:01

• 이 구조는 DB 수가 많아도 확장성이 좋습니다.
• 필요 시 try-except로 개별 DB 실패에 따른 예외 처리를 분리할 수 있습니다.
• pools는 전역 상태로 관리할 수 있으므로 이후에도 재사용이 가능합니다.