Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

發布時間：2020-08-30 20:14:46 來源：腳本之家閱讀：336 作者：基基偉欄目：開發技術

Oracle Client 安裝

1、環境

日期：2019年8月1日

公司已經安裝好Oracle服務端

Windows版本：Windows10專業版

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

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

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

2、下載網址

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

3、解壓至目錄

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

解壓后（這里放D盤）

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

4、配置環境變量

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

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

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

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

編輯PATH，添加%ORACLE_HOME%

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

Navicat連接測試

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

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, 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.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, table, 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, table, field, pattern, regexp=True, clob=True):
  sql = "SELECT %s FROM %s WHERE INSTR(%s,'%s')>0" % (field, table, field, 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之環境配置、實例代碼及報錯解決方法請查看下面的相關鏈接

向AI問一下細節

中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

1、環境

2、下載網址

3、解壓至目錄

4、配置環境變量

cx_Oracle

猜你喜歡

中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

Python連接Oracle之環境配置、實例代碼及報錯解決方法詳解

1、環境

2、下載網址

3、解壓至目錄

4、配置環境變量

cx_Oracle

猜你喜歡

最新資訊

相關推薦

相關標簽