Python连接Oracle之环境配置、实例代码及报错解决方法详解

2023-08-01 18:34:04 62

OracleClient安装

1、环境

日期：2019年8月1日

公司已经安装好Oracle服务端

Windows版本：Windows10专业版

系统类型：64位操作系统，基于x64的处理器

Python版本：Python3.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

安装命令

condainstallcx_Oracle

基础代码

importcx_Oracle
defexecute(query):
db=cx_Oracle.connect('用户名/密码@IP/ServiceName')
cursor=db.cursor()
cursor.execute(query)
result=cursor.fetchall()
cursor.close()
db.close()
returnresult
defcommit(sql):
db=cx_Oracle.connect('用户名/密码@IP/ServiceName')
cursor=db.cursor()
cursor.execute(sql)
db.commit()
cursor.close()
db.close()

封装成类

fromcx_OracleimportConnection#condainstallcx_Oracle
fromconfimportCONN,Color
classOracle(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()
defcommit(self,sql):
try:
self.cursor.execute(sql)
self.db.commit()
exceptExceptionase:
self.red(e)
deffetchall(self,query):
self.cursor.execute(query)
returnself.cursor.fetchall()
deffetchone(self,query,n=9999999):
self.cursor.execute(query)
for_inrange(n):
one=self.cursor.fetchone()
ifone:
yieldone
deffetchone_dt(self,query,n=9999999):
self.cursor.execute(query)
columns=[i[0]foriinself.cursor.description]
length=len(columns)
for_inrange(n):
one=self.cursor.fetchone()#tuple
yield{columns[i]:one[i]foriinrange(length)}
defread_clob(self,query):
self.cursor.execute(query)
one=self.cursor.fetchone()
whileone:
try:
yieldone[0].read()
exceptExceptionase:
self.red(e)
one=self.cursor.fetchone()
defdb2sheet(self,query,prefix):
df=pd.read_sql_query(query,self.db)
if'url'indf.columns:
df['url']="'"+df['url']
df.to_excel(prefix.replace('.xlsx','')+'.xlsx',index=False)
defdb2sheets(self,queries,prefix):
writer=pd.ExcelWriter(prefix.replace('.xlsx','')+'.xlsx')
forsheet_name,queryinqueries.items():
df=pd.read_sql_query(query,self.db)
if'url'indf.columns:
df['url']="'"+df['url']
df.to_excel(writer,sheet_name=sheet_name,index=False)
writer.save()
deftb2sheet(self,table):
sql="SELECT*FROM"+table
self.db2sheet(sql,table)
definsert(self,dt,tb):
fork,vindt.items():
ifisinstance(v,str):
dt[k]=v.replace("'",'').strip()
ls=[(k,v)fork,vindt.items()ifvisnotNone]
sql='INSERTINTO%s('%tb+','.join(i[0]foriinls)+\
')VALUES('+','.join('%r'%i[1]foriinls)+')'
self.commit(sql)
definsert_clob(self,dt,tb,clob):
fork,vindt.items():
ifisinstance(v,str):
dt[k]=v.replace("'",'').strip()
#把超长文本保存在一个变量中
#declare="DECLAREvariateCLOB:='%s';\n"%dt[clob]
join=lambdax:'||'.join("'%s'"%x[10922*i:10922*(i+1)]foriinrange(len(x)//10922+1))#32768//3
declare="DECLAREvariateCLOB:=%s;\n"%join(dt[clob])
dt[clob]='variate'
ls=[(k,v)fork,vindt.items()ifvisnotNone]
sql='INSERTINTO%s('%tb+','.join(i[0]foriinls)+')VALUES('+\
','.join('%r'%i[1]foriinls)+');'
sql=declare+'BEGIN\n%s\nEND;'%sql.replace("'variate'",'variate')
self.commit(sql)
defupdate(self,dt_update,dt_condition,table):
sql='UPDATE%sSET'%table+','.join('%s=%r'%(k,v)fork,vindt_update.items())\
+'WHERE'+'AND'.join('%s=%r'%(k,v)fork,vindt_condition.items())
self.commit(sql)
deftruncate(self,tb):
self.commit('truncatetable'+tb)
db_read=Oracle()
fetchall=db_read.fetchall
fetchone=db_read.fetchone
read_clob=db_read.read_clob
if__name__=='__main__':
query='''
'''.strip()
foriinfetchone(query,99):
print(i)

conf

CONN=('用户名','密码','IP/ServiceName')
conn='用户名/密码@IP/ServiceName'

文本字符串查询

classINSTR(Oracle):
"""文本字符串查询"""
defhighlight_instr(self,table,field,keyword,clob=True):
sql="SELECT%sFROM%sWHEREINSTR(%s,'%s')>0"%(field,table,field,keyword)
ifclob:
foriinself.read_clob(sql):
self.highlight(i,keyword)
else:
fori,inself.fetchone(sql):
self.highlight(i,keyword)
defregexp_instr(self,table,field,pattern,regexp=True,clob=True):
sql="SELECT%sFROM%sWHEREINSTR(%s,'%s')>0"%(field,table,field,pattern)
sql=sql.replace('INSTR','REGEXP_INSTR')ifregexpelsesql
ifclob:
foriinself.read_clob(sql):
yieldi
else:
fori,inself.fetchone(sql):
yieldi

一个简单的建表示例

--建表
CREATETABLEtable_name
(
serial_numberNUMBER(10),
collect_dateDATE,
urlVARCHAR2(255),
long_textCLOB,
priceNUMBER(10)--若需要精确到小数点2位，按分存储，/100还原到元
);
--给表添加备注
COMMENTONTABLEtable_nameIS'中文表名';
--给表字段添加备注
COMMENTONCOLUMNtable_name.serial_numberIS'编号';
COMMENTONCOLUMNtable_name.collect_dateIS'日期';
COMMENTONCOLUMNtable_name.urlIS'URL';
COMMENTONCOLUMNtable_name.long_textIS'长文本';
COMMENTONCOLUMNtable_name.priceIS'价钱';
--插入
INSERTINTOtable_name(collect_date)VALUES(DATE'2019-08-23');
INSERTINTOtable_name(long_text)VALUES('a');
INSERTINTOtable_name(long_text)VALUES('b');
--查询
SELECT*FROMtable_nameWHERETO_CHAR(long_text)in('a','b');
--查建表语句（表名大写）
SELECTdbms_metadata.get_ddl('TABLE','TABLE_NAME')FROMdual;
--删表
DROPTABLEtable_name;

sqlalchemy

importos#解决【UnicodeEncodeError:'ascii'codeccan'tencodecharacter】问题
os.environ['NLS_LANG']='AMERICAN_AMERICA.AL32UTF8'
#os.environ['NLS_LANG']='SIMPLIFIEDCHINESE_CHINA.UTF8'
fromcx_Oracleimportmakedsn
fromsqlalchemyimportcreate_engine,Column,String,Integer
fromsqlalchemy.ext.declarativeimportdeclarative_base
fromsqlalchemy.ormimportsessionmaker
#连接数据库（ORA-12505:TNS:listenerdoesnotcurrentlyknowofSIDgiveninconnectdescriptor）
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()
classStudent(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-bitOracleClientlibrarycannotbeloaded

首先操作系统位数、python位数、cx_Oracle版本要对应上；另外可能缺【VisualC++】

每次装完后，要重启pycharm和python

ORA-12170:TNS:Connecttimeoutoccurred

打开终端ping一下

检查【主机名或IP地址】、【服务名或SID】、【用户名】和【密码】是否填对

中文乱码

encoding=‘utf8'

ORA-00972:identifieristoolong

insert语句中出现'之类的字符

解决方法：将可能报错的字符替换掉

ORA-64203:DestinationbuffertoosmalltoholdCLOBdataaftercharactersetconversion.

selectTO_CHAR(long_text)fromtable_name，目标缓冲区太小，无法储存CLOB转换字符后的数据

解决方法：不在SQL用TO_CHAR，改在Python中用read（如上代码所示）

ORA-01704:stringliteraltoolong

虽然CLOB可以保存长文本，但是SQL语句有长度限制

解决方法：把超长文本保存在一个变量中（如上代码所示）

PLS-00172:stringliteraltoolong

字符串长度>32767(215-1)

解决方法：使用'||'来连接字符串（如上代码所示）

ORA-00928:missingSELECTkeyword

INSERT操作时，表字段命名与数据库内置名称冲突，如：ID、LEVEL、DATE等

解决方法：建立命名规范

cx_Oracle.DatabaseError:ORA-12505:TNS:listenerdoesnotcurrentlyknowofSIDgiveninconnectdescriptor

使用sqlalchemy时的报错

原因可能是目标数据库是集群部署的，可以咨询一下DBA，或见上面代码fromcx_Oracleimportmakedsn

UnicodeEncodeError:'ascii'codeccan'tencodecharacter

使用sqlalchemy时的报错，插入中文字符引起

解决方法是设置os.environ['NLS_LANG']

更多关于Python连接Oracle之环境配置、实例代码及报错解决方法请查看下面的相关链接

声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。