2025년 6월 23일 (월) 23:58에 Devcafe님의 편집

2025-06-23T23:58:52Z

← 이전 판		2025년 6월 24일 (화) 08:58 판
1번째 줄:		1번째 줄:
	~~아래는 요청하신 내용을 반영한 OracleConnectionManager 클래스 개선 버전입니다.~~

	~~✅ 변경 사항 요약~~		== 파이썬 오라클 커넥션 풀 확장 ==
	• get_connection_with_retry() 메서드를 클래스 내부로 통합		* get_connection_with_retry() 메서드를 클래스 내부로 통합
	• fetch_data() 메서드 추가 (SELECT 결과 반환)		* fetch_data() 메서드 추가 (SELECT 결과 반환)
	• 전체 구조 정돈 및 예외처리 강화		* 전체 구조 정돈 및 예외처리 강화

	⸻

	~~📦 oracle_connection_manager.py~~

			=== oracle_connection_manager.py ===
			<source lang=python>
	import json		import json
	import time		import time
46번째 줄:		44번째 줄:
	"""Get or acquire a new connection with retry logic."""		"""Get or acquire a new connection with retry logic."""
	if self.connections[db_name] is not None:		if self.connections[db_name] is not None:
	print(f"🔄 Using existing connection for {db_name}")		print(f" Using existing connection for {db_name}")
	return self.connections[db_name]		return self.connections[db_name]

54번째 줄:		52번째 줄:
	conn = self.pools[db_name].acquire()		conn = self.pools[db_name].acquire()
	self.connections[db_name] = conn		self.connections[db_name] = conn
	print(f"✅ Connection acquired on attempt {attempt + 1} for {db_name}")		print(f" Connection acquired on attempt {attempt + 1} for {db_name}")
	return conn		return conn
	except cx_Oracle.DatabaseError as e:		except cx_Oracle.DatabaseError as e:
	err, = e.args		err, = e.args
	print(f"⚠️ Attempt {attempt + 1} failed for {db_name}: {err.message}")		print(f" Attempt {attempt + 1} failed for {db_name}: {err.message}")
	attempt += 1		attempt += 1
	time.sleep(self.retry_delay)		time.sleep(self.retry_delay)
	raise Exception(f"❌ Unable to acquire DB connection for {db_name} after {self.max_retries} attempts.")		raise Exception(f" Unable to acquire DB connection for {db_name} after {self.max_retries} attempts.")

	def release_connection(self, db_name):		def release_connection(self, db_name):
68번째 줄:		66번째 줄:
	self.pools[db_name].release(self.connections[db_name])		self.pools[db_name].release(self.connections[db_name])
	self.connections[db_name] = None		self.connections[db_name] = None
	print(f"🔓 Connection released for {db_name}")		print(f" Connection released for {db_name}")

	def fetch_data(self, db_name, query, params=None):		def fetch_data(self, db_name, query, params=None):
86번째 줄:		84번째 줄:
	self.release_connection(db_name)		self.release_connection(db_name)
	self.pools[db_name].close()		self.pools[db_name].close()
	print(f"🛑 Pool closed for {db_name}")		print(f" Pool closed for {db_name}")

			</source>


	⸻		=== 사용 예시 ===
			<source lang=python>
	🧪 사용 예시

	if __name__ == "__main__":		if __name__ == "__main__":
103번째 줄:		101번째 줄:
	finally:		finally:
	manager.close_all()		manager.close_all()
			</source>

			=== 추가 개선 사항 ===
			* execute_update() 메서드 추가 (INSERT, UPDATE 등 트랜잭션 처리 포함)
			* asyncio 기반 비동기 구조 전환
			* 커넥션 상태 모니터링 (예: 자동 재연결 감지)

			[[category:python]]
	⸻		[[category:database]]
			[[category:sql]]
	~~📌 추가 개선 제안 (원하시면 확장 가능)~~
	~~• execute_update() 메서드 추가 (INSERT, UPDATE 등 트랜잭션 처리 포함)~~
	~~• asyncio 기반 비동기 구조 전환~~
	~~• 커넥션 상태 모니터링 (예~~: ~~자동 재연결 감지)~~

	~~필요하신 방향(비동기 처리, 트랜잭션 묶기 등)을 알려주시면 추가로 구성해드릴 수 있어요.~~

Devcafe: 새 문서: 아래는 요청하신 내용을 반영한 OracleConnectionManager 클래스 개선 버전입니다. ✅ 변경 사항 요약 • get_connection_with_retry() 메서드를 클래스 내부로 통합 • fetch_data() 메서드 추가 (SELECT 결과 반환) • 전체 구조 정돈 및 예외처리 강화 ⸻ 📦 oracle_connection_manager.py import json import time import cx_Oracle class OracleConnectionManager: def init(self, config_file, max_retries=3, retry_dela...

2025-05-28T10:46:37Z

새 문서: 아래는 요청하신 내용을 반영한 OracleConnectionManager 클래스 개선 버전입니다. ✅ 변경 사항 요약 • get_connection_with_retry() 메서드를 클래스 내부로 통합 • fetch_data() 메서드 추가 (SELECT 결과 반환) • 전체 구조 정돈 및 예외처리 강화 ⸻ 📦 oracle_connection_manager.py import json import time import cx_Oracle class OracleConnectionManager: def __init__(self, config_file, max_retries=3, retry_dela...

새 문서

아래는 요청하신 내용을 반영한 OracleConnectionManager 클래스 개선 버전입니다.

✅ 변경 사항 요약
• get_connection_with_retry() 메서드를 클래스 내부로 통합
• fetch_data() 메서드 추가 (SELECT 결과 반환)
• 전체 구조 정돈 및 예외처리 강화

⸻

📦 oracle_connection_manager.py

import json
import time
import cx_Oracle

class OracleConnectionManager:
def __init__(self, config_file, max_retries=3, retry_delay=2):
self.pools = {}
self.connections = {}
self.max_retries = max_retries
self.retry_delay = retry_delay
self.load_config(config_file)
self.create_pools()

def load_config(self, config_file):
with open(config_file, 'r') as file:
self.db_config = json.load(file)["Database"]

def create_pools(self):
for db in self.db_config:
dsn = cx_Oracle.makedsn(db["Host"], db["Port"], service_name=db["ServiceName"])
pool = cx_Oracle.SessionPool(
user=db["Username"],
password=db["Password"],
dsn=dsn,
min=1,
max=5,
increment=1,
threaded=True,
getmode=cx_Oracle.SPOOL_ATTRVAL_WAIT
)
self.pools[db["Name"]] = pool
self.connections[db["Name"]] = None

def get_connection(self, db_name):
"""Get or acquire a new connection with retry logic."""
if self.connections[db_name] is not None:
print(f"🔄 Using existing connection for {db_name}")
return self.connections[db_name]

attempt = 0
while attempt < self.max_retries:
try:
conn = self.pools[db_name].acquire()
self.connections[db_name] = conn
print(f"✅ Connection acquired on attempt {attempt + 1} for {db_name}")
return conn
except cx_Oracle.DatabaseError as e:
err, = e.args
print(f"⚠️ Attempt {attempt + 1} failed for {db_name}: {err.message}")
attempt += 1
time.sleep(self.retry_delay)
raise Exception(f"❌ Unable to acquire DB connection for {db_name} after {self.max_retries} attempts.")

def release_connection(self, db_name):
"""Release the connection back to the pool."""
if self.connections[db_name]:
self.pools[db_name].release(self.connections[db_name])
self.connections[db_name] = None
print(f"🔓 Connection released for {db_name}")

def fetch_data(self, db_name, query, params=None):
"""Execute SELECT query and return results."""
conn = self.get_connection(db_name)
cursor = conn.cursor()
try:
cursor.execute(query, params or {})
return cursor.fetchall()
finally:
cursor.close()

def close_all(self):
"""Close all pools and connections."""
for db_name in self.pools:
if self.connections[db_name]:
self.release_connection(db_name)
self.pools[db_name].close()
print(f"🛑 Pool closed for {db_name}")

⸻

🧪 사용 예시

if __name__ == "__main__":
manager = OracleConnectionManager("db_config.json")

try:
rows = manager.fetch_data("DB1", "SELECT empno, ename FROM emp WHERE deptno = :dept", {"dept": 10})
for row in rows:
print(row)
finally:
manager.close_all()

⸻

📌 추가 개선 제안 (원하시면 확장 가능)
• execute_update() 메서드 추가 (INSERT, UPDATE 등 트랜잭션 처리 포함)
• asyncio 기반 비동기 구조 전환
• 커넥션 상태 모니터링 (예: 자동 재연결 감지)

필요하신 방향(비동기 처리, 트랜잭션 묶기 등)을 알려주시면 추가로 구성해드릴 수 있어요.

파이썬 오라클 커넥션 풀 - 편집 역사

2025년 6월 23일 (월) 23:58에 Devcafe님의 편집