Devcafe: 새 문서: == 아키텍처 == https://oracle.github.io/python-cx_Oracle/samples/tutorial/resources/cx_Oracle_arch.png ---- == 설치 / 설정 == === 접속 모듈 설치 === pip install cx_Oracle ---- === 사용자 매뉴얼 === https://cx-oracle.readthedocs.io/en/latest/user_guide/sql_execution.html#fetch-methods == 연결 == === 접속 테스트 === import cx_Oracle #한글 지원 방법 import os os.putenv('NLS_LANG', '.UTF8') #연...

2024-10-04T15:25:29Z

새 문서: == 아키텍처 == https://oracle.github.io/python-cx_Oracle/samples/tutorial/resources/cx_Oracle_arch.png ---- == 설치 / 설정 == === 접속 모듈 설치 === <source lang=python> pip install cx_Oracle </source> ---- === 사용자 매뉴얼 === https://cx-oracle.readthedocs.io/en/latest/user_guide/sql_execution.html#fetch-methods == 연결 == === 접속 테스트 === <source lang=python> import cx_Oracle #한글 지원 방법 import os os.putenv('NLS_LANG', '.UTF8') #연...

새 문서

== 아키텍처 ==
https://oracle.github.io/python-cx_Oracle/samples/tutorial/resources/cx_Oracle_arch.png
----
== 설치 / 설정 ==
=== 접속 모듈 설치 ===
<source lang=python>
pip install cx_Oracle
</source>
----
=== 사용자 매뉴얼 ===
https://cx-oracle.readthedocs.io/en/latest/user_guide/sql_execution.html#fetch-methods

== 연결 ==
=== 접속 테스트 ===
<source lang=python>
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()
</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>

=== DRCP 커넥션풀 사용하기 ===

{{고지상자
|제목 = DRCP (Database Resident Connection Pooling)
|내용 = # DRCP (Database Resident Connection Pooling) 기능은 DB 접속을 위한 커넥션 공유풀을 공동사용하게 함으로써 DB 서버의 자원 사용을 절약하는 기능
# 만약 1000개의 프로세스가 DB 접속 하는 경우, 기본적으로 1000개의 클라이언트 처리요청을 지원하기 위해 1000개의 Dedicated Server Process 를 기동 해야함
# WAS 를 사용하지 않는 경우, DB 차원에서 이런 "공유 커넥션 풀" 기능을 제공
# 오라클 DB에서 2가지 Shared Server 방식과 DRCP(Database Resident Connection Pooling) 방식 제공
}}
* 동시 접속이 많은 경우 DB 서버의 자원이 금방 고갈될 수 밖에 없는 문제를 해결하기 위해 보통은 WAS (Web Application Server) 차원에서 커넥션풀을 만들어서 DB접속풀 공유하면서 사용하는 것이 일반적임.
<source lang=python>

import cx_Oracle
import threading

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

def Query():
con = pool.acquire(cclass = "PYTHONHOL", purity = cx_Oracle.ATTR_PURITY_SELF)
cur = conn.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>

== 조회 / 패치 ==
=== fetchone() 함수를 사용하여 1건 패치 ===
* 로우갯수가 많을때 fetchall() 함수는 너무 많은 메모리

<source lang=python>
import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()

cur.execute("select * from dept order by deptno")
row = cur.fetchone()
print(row)

row = cur.fetchone()
print(row)
</source>

=== fetchmany() 사용하여 다건 패치 ===

<source lang=python>
import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()

cur.execute("select * from dept order by deptno")
res = cur.fetchmany(numRows = 3)
print(res)
</source>

=== Scrollable cursors ===
# 스크롤 가능한 커서를 사용하면 응용 프로그램이 쿼리 결과에서 앞뒤로 이동할 수 있습니다.
# 행을 건너뛰고 특정 행으로 이동하는 데 사용할 수 있습니다.
<source lang=python>
import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor(scrollable = True)

cur.execute("select * from dept order by deptno")

cur.scroll(2, mode = "absolute") # go to second row
print(cur.fetchone())

cur.scroll(-1) # go back one row
print(cur.fetchone())

</source>

<source lang=python>
cur.scroll(1) # go to next row
print(cur.fetchone())

cur.scroll(mode = "first") # go to first row
print(cur.fetchone())
</source>

=== 프리패치 ===
# Python 프로그램으로 각 일괄 처리에서 반환되는 행 수를 늘려 쿼리 성능을 개선하는 방법
# 행을 미리 가져오기 및 배열 가져오기는 모두 데이터베이스로의 왕복을 줄이기 위한 내부 버퍼링 기술입니다.
# 차이점은 버퍼링을 수행하는 코드 레이어와 버퍼링이 발생하는 시기입니다.
<source lang=python>
import cx_Oracle
import time
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)

start = time.time()

cur = con.cursor()
cur.prefetchrows = 100
cur.arraysize = 100
cur.execute("select * from bigtab")
res = cur.fetchall()
# print(res) # uncomment to display the query results

elapsed = (time.time() - start)
print(elapsed, "seconds")
</source>

=== 바인드 변수 사용하기 ===
<source lang=python>

import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()

sql = "select * from dept where deptno = :id order by deptno"

cur.execute(sql, id = 20)
res = cur.fetchall()
print(res)

cur.execute(sql, id = 10)
res = cur.fetchall()
print(res)
</source>

== 바인딩 입력 처리 ==
<source lang=python>

import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()

rows = [ (1, "First" ), (2, "Second" ),
(3, "Third" ), (4, "Fourth" ),
(5, "Fifth" ), (6, "Sixth" ),
(7, "Seventh" ) ]

cur.executemany("insert into mytab(id, data) values (:1, :2)", rows)

# Now query the results back

cur2 = con.cursor()
cur2.execute('select * from mytab')
res = cur2.fetchall()
print(res)
</source>

== 에러 처리 ==
=== 배치 에러 ===
<source lang=python>
import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()

# 예를 들어 6 이 중복입력 되어 유니크 제약 조건 오류가 발생된경우

rows = [ (1, "First" ), (2, "Second" ),
(3, "Third" ), (4, "Fourth" ),
(5, "Fifth" ), (6, "Sixth" ),
(6, "Duplicate" ),
(7, "Seventh" ) ]

cur.executemany("insert into mytab(id, data) values (:1, :2)", rows, batcherrors = True)

for error in cur.getbatcherrors():
print("Error", error.message.rstrip(), "at row offset", error.offset)

# Now query the results back

cur2 = con.cursor()
cur2.execute('select * from mytab')
res = cur2.fetchall()
print(res)

</python>
The other data gets inserted and is queried back.
에러가 발생된 값을 제외 하고 모두 입력된다.

또한 커밋 이나 롤백 처리를 할수 있다.

* 커밋처리 시
<source lang=sql>
con.commit()
</source>

* 롤백처리 시
<source lang=sql>
con.rollback()
</source>

== LOB 처리 ==
=== CLOB 조회 ===
<source lang=python>
import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()

print("Inserting data...")
cur.execute("truncate table testclobs")
longString = ""
for i in range(5):
char = chr(ord('A') + i)
longString += char * 250
cur.execute("insert into testclobs values (:1, :2)",
(i + 1, "String data " + longString + ' End of string'))
con.commit()

print("Querying data...")
cur.execute("select * from testclobs where id = :id", {'id': 1})
(id, clob) = cur.fetchone()
print("CLOB length:", clob.size())
clobdata = clob.read()
print("CLOB data:", clobdata)

</source>

=== CLOB 문자열 처리 ===
<source lang=python>
import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()

print("Inserting data...")
cur.execute("truncate table testclobs")
longString = ""
for i in range(5):
char = chr(ord('A') + i)
longString += char * 250
cur.execute("insert into testclobs values (:1, :2)",
(i + 1, "String data " + longString + ' End of string'))
con.commit()

def OutputTypeHandler(cursor, name, defaultType, size, precision, scale):
if defaultType == cx_Oracle.CLOB:
return cursor.var(cx_Oracle.LONG_STRING, arraysize = cursor.arraysize)

con.outputtypehandler = OutputTypeHandler

print("Querying data...")
cur.execute("select * from testclobs where id = :id", {'id': 1})
(id, clobdata) = cur.fetchone()
print("CLOB length:", len(clobdata))
print("CLOB data:", clobdata)

</source>

* 참조 https://oracle.github.io/python-cx_Oracle/samples/tutorial/Python-and-Oracle-Database-Scripting-for-the-Future.html

[[Category:python]]

Python 오라클 연결 cx oracle - 편집 역사