Python에서 Oracle 데이터베이스를 운영하기 위한 간단한 메서드 및 캡슐화된 클래스 예제

이 글에서는 오라클 데이터베이스를 운영하기 위한 Python의 간단한 메소드와 캡슐화 클래스를 주로 소개하며, Python의 간단한 연결, 쿼리, Oracle 데이터베이스 닫기의 기본 작업을 예제 형식으로 분석하고 다양한 Oracle 작업에 대한 Python 캡슐화 클래스를 제공합니다. 필요하신 분들은 참고하세요

이 글은 Python에서 오라클 데이터베이스를 운영하기 위한 간단한 메소드와 캡슐화 클래스를 설명합니다. 참고하실 수 있도록 자세한 내용을 공유해드리고자 합니다.

최근 직장에서 오라클을 접하게 되었는데, Python 스크립트를 여러 곳에서 사용하는 것이 훨씬 편리할 것 같다는 생각이 들었습니다. 먼저 Python에서 Oracle을 작동하는 기본 방법을 알아봅니다.

Oracle의 사용과 OracleClient의 NetConfig 존재를 고려하면, 연결하는 것이 간단한 문제가 되어서는 안된다고 생각합니다.

물론 인터넷에서 여러 연결 방법을 찾아 한참을 그렸으나 특종을 찾지 못했습니다.

방법 1: 사용자 이름, 비밀번호, 모니터링을 각각 매개 변수로 사용합니다

conn=cx_Oracle.connect('用户名','密码','数据库地址:数据库端口/SID')

제가 읽은 여러 기사와 코드 작성 오류에 대한 팁에 따르면 Python에 연결할 구성 항목이 있음을 발견했습니다. 데이터베이스는 Oracle과 동일해야 합니다. 클라이언트 구성 파일 tnsnames.ora가 관련되어 있습니다. 그런데 제가 만든 구성 항목에는 SID 항목이 없고, 처음에는 SID가 무엇인지도 모르고 그냥 인터넷에 나온 대로 따라해 봤기 때문에 이 방법은 실패했습니다. 나중에 구성 항목에 SID를 추가해야 한다는 사실을 알게 되었고, 이 항목을 구성한 후 시스템을 다시 시작해야 하는지 생각해 보았습니다. 그럼 먼저 다른 방법들을 살펴보겠습니다...

방법 2: 사용자 이름, 비밀번호, 모니터링을 하나의 매개변수로 사용합니다

conn=cx_Oracle.connect('用户名/密码@数据库地址:数据库端口/SID')

이 방법은 기본적으로 방법 1과 동일하지만 차이점이 있습니다. medicine...

방법 3: tns 구성 정보 사용

conn=cx_Oracle.connect('用户名','密码',tns)

tns를 얻기 위한 인터넷의 코드는 함수를 사용하여 수행되며 SID는 계속 사용되지만... 이미 SID가 없어도 사용할 수 있어서

tns=cx_Oracle.makedsn('数据库地址','数据库端口', 'SID')

을 사용해도 여전히 작동하지 않지만 이 tns의 생성 방법을 보면 위의 두 가지 방법과 유사합니다. 그런데 SID를 무작위로 입력하면 생성되는 데이터가 이렇다는 걸 알았습니다.

(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=127.0.0.1)(PORT=1521)))(CONNECT_DATA=(SID=XE)))

그러나 내 클라이언트의 구성은 항목은 아마도 다음과 같습니다.

(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=127.0.0.1)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=KGDB)))

보세요, 똑같아 보이는데, 유형이 모두 문자열 유형입니다. 내 파일의 구성 항목을 tns에 직접 할당해 보세요.

tns = '(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=127.0.0.1)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=KGDB)))'
conn = cx_Oracle.connect('nicker', '123456', tns)

흠. 성공~

마지막으로 기본 사용법의 전체 코드를 게시합니다

#coding:utf-8
import cx_Oracle
# 创建数据库连接
# cx_Oracle.connect('username','pwd','ora的tns信息')
# oracle数据库的tns信息，从tnsnames.ora中找到plsql可用的配置项，将该配置项直接拷贝过来即可
ora_tns = '(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=127.0.0.1)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=KGDB)))'
conn = cx_Oracle.connect('nicker', '123456', ora_tns)
# 操作游标
cursor = conn.cursor()
# 执行查询
cursor.execute("SELECT * FROM inst_info")
# 获取返回信息
rs = cursor.fetchall()
# 输出信息
for v in rs:
  print v
#关闭连接，释放资源
cursor.close()
conn.close()

관찰 및 발견 요약이 매우 중요하며 이해도 필요합니다

Oracle을 캡슐화하는 수업 게시

#coding:utf-8
import cx_Oracle
# 封装的类
class cxOracle:
  '''
  tns的取值tnsnames.ora对应的配置项的值，如：
  tns = '(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=10.16.18.23)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=MYDB)))'
  '''
  def __init__(self ,uname, upwd,tns ):
    self ._uname = uname
    self ._upwd = upwd
    self ._tns = tns
    self ._conn = None
    self ._ReConnect()
  def _ReConnect(self ):
    if not self._conn :
      self ._conn = cx_Oracle.connect (self. _uname, self ._upwd, self._tns)
    else:
      pass
  def __del__(self ):
    if self. _conn:
      self ._conn. close()
      self ._conn = None
  def _NewCursor(self ):
    cur = self. _conn.cursor ()
    if cur:
      return cur
    else:
      print "#Error# Get New Cursor Failed."
      return None
  def _DelCursor(self , cur):
    if cur:
      cur .close()
  # 检查是否允许执行的sql语句
  def _PermitedUpdateSql(self ,sql):
    rt = True
    lrsql = sql. lower()
    sql_elems = [ lrsql.strip ().split()]
    # update和delete最少有四个单词项
    if len( sql_elems) < 4 :
      rt = False
    # 更新删除语句，判断首单词，不带where语句的sql不予执行
    elif sql_elems[0] in [ &#39;update&#39;, &#39;delete&#39;]:
      if &#39;where&#39; not in sql_elems :
        rt = False
    return rt
  # 导出结果为文件
  def Export(self , sql, file_name, colfg =&#39;||&#39;):
    rt = self. Query(sql )
    if rt:
      with open( file_name, &#39;a&#39;) as fd:
        for row in rt:
          ln_info = &#39;&#39;
          for col in row:
             ln_info += str( col) + colfg
          ln_info += &#39;\n&#39;
          fd .write( ln_info)
  # 查询
  def Query(self , sql, nStart=0 , nNum=- 1):
    rt = []
    # 获取cursor
    cur = self. _NewCursor()
    if not cur:
      return rt
    # 查询到列表
    cur .execute(sql)
    if ( nStart==0 ) and (nNum==1 ):
      rt .append( cur.fetchone ())
    else:
      rs = cur. fetchall()
      if nNum==- 1:
        rt .extend( rs[nStart:])
      else:
        rt .extend( rs[nStart:nStart +nNum])
    # 释放cursor
    self ._DelCursor(cur)
    return rt
  # 更新
  def Exec(self ,sql):
    # 获取cursor
    rt = None
    cur = self. _NewCursor()
    if not cur:
      return rt
    # 判断sql是否允许其执行
    if not _PermitedUpdateSql(sql ):
      return rt
    # 执行语句
    rt = cur. execute(sql )
    # 释放cursor
    self ._DelCursor(cur)
    return rt
# 类使用示例
tns = &#39;(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=10.16.17.46)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=MYDB)))&#39;
ora = cxOracle (&#39;nicker&#39;, &#39;123456&#39;, tns)
# 导出结果为文件
rs = ora .Export("SELECT * FROM org", &#39;1.txt&#39;)
# 查询结果到列表
rs = ora.Query("SELECT * FROM org")
print rs
# 更新数据
ora.Exec("update org set org_name=&#39;NewNameForUpdate&#39; where org_id=123456;")

관련 권장 사항:

cx_Oracle 모듈을 사용하여 Oracle 데이터베이스를 운영하는 Python에 대한 자세한 설명

위 내용은 Python에서 Oracle 데이터베이스를 운영하기 위한 간단한 메서드 및 캡슐화된 클래스 예제의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!