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_Oracle
def execute(query):
  db = cx_Oracle.connect('使用者名稱/密碼@IP/ServiceName')
  cursor = db.cursor()
  cursor.execute(query)
  result = cursor.fetchall()
  cursor.close()
  db.close()
  return result
def 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_Oracle
from conf import CONN,Color
class 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,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,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,tb,clob):
    for k,'').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,'.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 in dt_update.items()) \
       + ' WHERE ' + ' AND '.join('%s=%r' % (k,v in dt_condition.items())
    self.commit(sql)
  def truncate(self,tb):
    self.commit('truncate table ' + tb)
db_read = Oracle()
fetchall = db_read.fetchall
fetchone = db_read.fetchone
read_clob = db_read.read_clob
if __name__ == '__main__':
  query = '''
  '''.strip()
  for i in fetchone(query,99):
    print(i)

conf

CONN = ('使用者名稱','密碼','IP/ServiceName')
conn = '使用者名稱/密碼@IP/ServiceName'

文字字串查詢

class INSTR(Oracle):
  """文字字串查詢"""
  def highlight_instr(self,table,field,keyword,clob=True):
    sql = "SELECT %s FROM %s WHERE INSTR(%s,'%s')>0" % (field,keyword)
    if clob:
      for i in self.read_clob(sql):
        self.highlight(i,keyword)
    else:
      for i,in self.fetchone(sql):
        self.highlight(i,keyword)
  def regexp_instr(self,pattern,regexp=True,pattern)
    sql = sql.replace('INSTR','REGEXP_INSTR') if regexp else sql
    if clob:
      for i in self.read_clob(sql):
        yield i
    else:
      for i,in self.fetchone(sql):
        yield i

一個簡單的建表示例

-- 建表
CREATE TABLE table_name
(
serial_number   NUMBER(10),collect_date   DATE,url        VARCHAR2(255),long_text     CLOB,price       NUMBER(10)-- 若需要精確到小數點2位，按分儲存，/100還原到元
);
-- 給表新增備註
COMMENT ON TABLE table_name IS '中文表名';
-- 給表字段新增備註
COMMENT ON COLUMN table_name.serial_number IS '編號';
COMMENT ON COLUMN table_name.collect_date IS '日期';
COMMENT ON COLUMN table_name.url IS 'URL';
COMMENT ON COLUMN table_name.long_text IS '長文字';
COMMENT ON COLUMN table_name.price IS '價錢';
-- 插入
INSERT INTO table_name(collect_date) VALUES (DATE'2019-08-23');
INSERT INTO table_name(long_text) VALUES ('a');
INSERT INTO table_name(long_text) VALUES ('b');
-- 查詢
SELECT * FROM table_name WHERE TO_CHAR(long_text) in ('a','b');
-- 查建表語句（表名大寫）
SELECT dbms_metadata.get_ddl('TABLE','TABLE_NAME') FROM dual;
-- 刪表
DROP TABLE table_name;

sqlalchemy

import os # 解決【UnicodeEncodeError: 'ascii' codec can't encode character】問題
os.environ['NLS_LANG'] = 'AMERICAN_AMERICA.AL32UTF8'
# os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
from cx_Oracle import makedsn
from sqlalchemy import create_engine,Column,String,Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 連線資料庫（ORA-12505: TNS:listener does not currently know of SID given in connect descriptor）
ip = ''
port = ''
tnsname = '' # 例項名
uname = '' # 使用者名稱
pwd = '' # 密碼
dsnStr = makedsn(ip,port,service_name=tnsname)
connect_str = "oracle://%s:%s@%s" % (uname,pwd,dsnStr)
# 建立連線引擎，這個engine是lazy模式，直到第一次被使用才真實建立
engine = create_engine(connect_str,encoding='utf-8')
# 建立物件的基類
Base = declarative_base()
class Student(Base):
  # 表名
  __tablename__ = 'student'
  # 表字段
  sid = Column(String(20),primary_key=True)
  age = Column(Integer)
# 建表（繼承Base的所有表）
Base.metadata.create_all(bind=engine)
# 使用ORM操作資料庫
Session = sessionmaker(bind=engine) # 建立ORM基類
session = Session() # 建立ORM物件
tb_obj = Student(sid='a6',age=18) # 建立表物件
session.add(tb_obj) # 新增到ORM物件（插入資料）
session.commit() # 提交
session.close() # 關閉ORM物件
# 刪表（繼承Base的所有表）
Base.metadata.drop_all(engine)

報錯處理

DPI-1047: 64-bit Oracle Client library cannot be loaded

首先作業系統位數、python位數、cx_Oracle版本要對應上；另外可能缺【Visual C++】

每次裝完後，要重啟pycharm和python

ORA-12170: TNS:Connect timeout occurred

開啟終端ping一下

檢查【主機名或IP地址】、【服務名或SID】、【使用者名稱】和【密碼】是否填對

中文亂碼

encoding=‘utf8'

ORA-00972: identifier is too long

insert語句中出現'之類的字元

解決方法：將可能報錯的字元替換掉

ORA-64203: Destination buffer too small to hold CLOB data after character set conversion.

select TO_CHAR(long_text) from table_name，目標緩衝區太小，無法儲存CLOB轉換字元後的資料

解決方法：不在SQL用TO_CHAR，改在Python中用read（如上程式碼所示）

ORA-01704: string literal too long

雖然CLOB可以儲存長文字，但是SQL語句有長度限制

解決方法：把超長文字儲存在一個變數中（如上程式碼所示）

PLS-00172: string literal too long

字串長度>32767(215-1)

解決方法：使用'||'來連線字串（如上程式碼所示）

ORA-00928: missing SELECT keyword

INSERT操作時，表字段命名與資料庫內建名稱衝突，如：ID、LEVEL、DATE等

解決方法：建立命名規範

cx_Oracle.DatabaseError: ORA-12505: TNS:listener does not currently know of SID given in connect descriptor

使用sqlalchemy時的報錯

原因可能是目標資料庫是叢集部署的，可以諮詢一下DBA，或見上面程式碼from cx_Oracle import makedsn

UnicodeEncodeError: 'ascii' codec can't encode character

使用sqlalchemy時的報錯，插入中文字元引起

解決方法是設定os.environ['NLS_LANG']

更多關於Python連線Oracle之環境配置、例項程式碼及報錯解決方法請檢視下面的相關連結