행위

"Python 오라클 연결 cx oracle"의 두 판 사이의 차이

DB CAFE

(접속 테스트)
(접속 테스트)
43번째 줄: 43번째 줄:
 
connection.close()
 
connection.close()
 
</source>
 
</source>
 +
 +
=== 커넥션 풀 사용하기 ===
 +
<source lang=python>
 +
import cx_Oracle
 +
import threading
 +
import db_config
 +
 +
pool = cx_Oracle.SessionPool(db_config.user, db_config.pw, db_config.dsn,
 +
                            min = 2, max = 5, increment = 1, threaded = True,
 +
                            getmode = cx_Oracle.SPOOL_ATTRVAL_WAIT)
 +
 +
def Query():
 +
    con = pool.acquire()
 +
    cur = con.cursor()
 +
    for i in range(4):
 +
        cur.execute("select myseq.nextval from dual")
 +
        seqval, = cur.fetchone()
 +
        print("Thread", threading.current_thread().name, "fetched sequence =", seqval)
 +
 +
numberOfThreads = 2
 +
threadArray = []
 +
 +
for i in range(numberOfThreads):
 +
    thread = threading.Thread(name = '#' + str(i), target = Query)
 +
    threadArray.append(thread)
 +
    thread.start()
 +
 +
for t in threadArray:
 +
    t.join()
 +
 +
print("All done!")
 +
 +
</source>
 +
  
 
[[Category:python]]
 
[[Category:python]]

2023년 5월 31일 (수) 15:07 판

thumb_up 추천메뉴 바로가기


1 개요[편집]

cx_Oracle_arch.png


2 접속 모듈 설치[편집]

pip install cx_Oracle

2.2 접속 테스트[편집]

import cx_Oracle

#한글 지원 방법
import os
os.putenv('NLS_LANG', '.UTF8')

#연결에 필요한 기본 정보 (유저, 비밀번호, 데이터베이스 서버 주소)
connection = cx_Oracle.connect('Id','password','localhost/orcl')

# 오라클 버전 확인 
print("Database version:", connection.version)
print("Client version:", cx_Oracle.clientversion())

# 커셔 연결 
cursor = connection.cursor()

cursor.execute("""
   select name
   from test_db
   where text = :texting""",
   texting = "테스트"
)

for name in cursor:
   print("테스트 이름 리스트 : ", name)


# 종료
cursor.close()
connection.close()

2.3 커넥션 풀 사용하기[편집]

import cx_Oracle
import threading
import db_config

pool = cx_Oracle.SessionPool(db_config.user, db_config.pw, db_config.dsn,
                             min = 2, max = 5, increment = 1, threaded = True,
                             getmode = cx_Oracle.SPOOL_ATTRVAL_WAIT)

def Query():
    con = pool.acquire()
    cur = con.cursor()
    for i in range(4):
        cur.execute("select myseq.nextval from dual")
        seqval, = cur.fetchone()
        print("Thread", threading.current_thread().name, "fetched sequence =", seqval)

numberOfThreads = 2
threadArray = []

for i in range(numberOfThreads):
    thread = threading.Thread(name = '#' + str(i), target = Query)
    threadArray.append(thread)
    thread.start()

for t in threadArray:
    t.join()

print("All done!")