Oracle Client 安裝

1、環境

日期：2019年8月1日

公司已經安裝好Oracle服務端

Windows版本：Windows10專業版

系統類型：64位操作系統，基于x64的處理器

Python版本：Python 3.6.4 :: Anaconda, Inc.

2、下載網址

https://www.oracle.com/database/technologies/instant-client/downloads.html

3、解壓至目錄

解壓后（這里放D盤）

4、配置環境變量

控制面板/系統和安全/系統 -> 高級系統設置 -> 環境變量

新建ORACLE_HOME，值為包解壓的路徑

編輯PATH，添加%ORACLE_HOME%

Navicat連接測試

cx_Oracle

安裝命令

conda install cx_Oracle

基礎代碼

import cx_Oracledef execute(query):  db = cx_Oracle.connect('用戶名/密碼@IP/ServiceName')  cursor = db.cursor()  cursor.execute(query)  result = cursor.fetchall()  cursor.close()  db.close()  return resultdef commit(sql):  db = cx_Oracle.connect('用戶名/密碼@IP/ServiceName')  cursor = db.cursor()  cursor.execute(sql)  db.commit()  cursor.close()  db.close()

封裝成類

from cx_Oracle import Connection # conda install cx_Oraclefrom conf import CONN, Colorclass Oracle(Color):  def __init__(self, conn=CONN):    self.db = Connection(*conn, encoding='utf8') # 用戶名 密碼 IP/ServiceName    self.cursor = self.db.cursor()  def __del__(self):    self.cursor.close()    self.db.close()  def commit(self, sql):    try:      self.cursor.execute(sql)      self.db.commit()    except Exception as e:      self.red(e)  def fetchall(self, query):    self.cursor.execute(query)    return self.cursor.fetchall()  def fetchone(self, query, n=9999999):    self.cursor.execute(query)    for _ in range(n):      one = self.cursor.fetchone()      if one:        yield one  def fetchone_dt(self, query, n=9999999):    self.cursor.execute(query)    columns = [i[0] for i in self.cursor.description]    length = len(columns)    for _ in range(n):      one = self.cursor.fetchone() # tuple      yield {columns[i]: one[i] for i in range(length)}  def read_clob(self, query):    self.cursor.execute(query)    one = self.cursor.fetchone()    while one:      try:        yield one[0].read()      except Exception as e:        self.red(e)      one = self.cursor.fetchone()  def db2sheet(self, query, prefix):    df = pd.read_sql_query(query, self.db)    if 'url' in df.columns:      df['url'] = "'" + df['url']    df.to_excel(prefix.replace('.xlsx', '')+'.xlsx', index=False)  def db2sheets(self, queries, prefix):    writer = pd.ExcelWriter(prefix.replace('.xlsx', '')+'.xlsx')    for sheet_name, query in queries.items():      df = pd.read_sql_query(query, self.db)      if 'url' in df.columns:        df['url'] = "'" + df['url']      df.to_excel(writer, sheet_name=sheet_name, index=False)    writer.save()  def tb2sheet(self, table):    sql = "SELECT * FROM " + table    self.db2sheet(sql, table)  def insert(self, dt, tb):    for k, v in dt.items():      if isinstance(v, str):        dt[k] = v.replace("'", '').strip()    ls = [(k, v) for k, v in dt.items() if v is not None]    sql = 'INSERT INTO %s (' % tb + ','.join(i[0] for i in ls) + /       ') VALUES (' + ','.join('%r' % i[1] for i in ls) + ')'    self.commit(sql)  def insert_clob(self, dt, tb, clob):    for k, v in dt.items():      if isinstance(v, str):        dt[k] = v.replace("'", '').strip()    # 把超長文本保存在一個變量中    # declare = "DECLARE variate CLOB := '%s';/n" % dt[clob]    join = lambda x: '||'.join("'%s'" % x[10922*i: 10922*(i+1)] for i in range(len(x)//10922+1)) # 32768//3    declare = "DECLARE variate CLOB := %s;/n" % join(dt[clob])    dt[clob] = 'variate'    ls = [(k, v) for k, v in dt.items() if v is not None]    sql = 'INSERT INTO %s (' % tb + ','.join(i[0] for i in ls) + ') VALUES (' +/       ','.join('%r' % i[1] for i in ls) + ');'    sql = declare + 'BEGIN/n%s/nEND;' % sql.replace("'variate'", 'variate')    self.commit(sql)  def update(self, dt_update, dt_condition, table):    sql = 'UPDATE %s SET ' % table + ','.join('%s=%r' % (k, v) for k, v in dt_update.items()) /       + ' WHERE ' + ' AND '.join('%s=%r' % (k, v) for k, v in dt_condition.items())    self.commit(sql)  def truncate(self, tb):    self.commit('truncate table ' + tb)db_read = Oracle()fetchall = db_read.fetchallfetchone = db_read.fetchoneread_clob = db_read.read_clobif __name__ == '__main__':  query = '''  '''.strip()  for i in fetchone(query, 99):    print(i)