电脑系统城官网 - 一键重装系统-Win7旗舰版/win10 64位系统下载,唯一官网:www.pcxitongcheng.com!

当前位置:首页 > 脚本中心 > python > 详细页面

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

时间:2020-02-12来源:系统城作者:电脑系统城

这篇文章主要介绍了Python连接Oracle之环境配置、实例代码及报错解决方法详解,需要的朋友可以参考下

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

安装命令


 
  1. conda install cx_Oracle

 

基础代码

 


 
  1. import cx_Oracle
  2. def execute(query):
  3. db = cx_Oracle.connect('用户名/密码@IP/ServiceName')
  4. cursor = db.cursor()
  5. cursor.execute(query)
  6. result = cursor.fetchall()
  7. cursor.close()
  8. db.close()
  9. return result
  10. def commit(sql):
  11. db = cx_Oracle.connect('用户名/密码@IP/ServiceName')
  12. cursor = db.cursor()
  13. cursor.execute(sql)
  14. db.commit()
  15. cursor.close()
  16. db.close()

 

封装成类

 


 
  1. from cx_Oracle import Connection # conda install cx_Oracle
  2. from conf import CONN, Color
  3. class Oracle(Color):
  4. def __init__(self, conn=CONN):
  5. self.db = Connection(*conn, encoding='utf8') # 用户名 密码 IP/ServiceName
  6. self.cursor = self.db.cursor()
  7. def __del__(self):
  8. self.cursor.close()
  9. self.db.close()
  10. def commit(self, sql):
  11. try:
  12. self.cursor.execute(sql)
  13. self.db.commit()
  14. except Exception as e:
  15. self.red(e)
  16. def fetchall(self, query):
  17. self.cursor.execute(query)
  18. return self.cursor.fetchall()
  19. def fetchone(self, query, n=9999999):
  20. self.cursor.execute(query)
  21. for _ in range(n):
  22. one = self.cursor.fetchone()
  23. if one:
  24. yield one
  25. def fetchone_dt(self, query, n=9999999):
  26. self.cursor.execute(query)
  27. columns = [i[0] for i in self.cursor.description]
  28. length = len(columns)
  29. for _ in range(n):
  30. one = self.cursor.fetchone() # tuple
  31. yield {columns[i]: one[i] for i in range(length)}
  32. def read_clob(self, query):
  33. self.cursor.execute(query)
  34. one = self.cursor.fetchone()
  35. while one:
  36. try:
  37. yield one[0].read()
  38. except Exception as e:
  39. self.red(e)
  40. one = self.cursor.fetchone()
  41. def db2sheet(self, query, prefix):
  42. df = pd.read_sql_query(query, self.db)
  43. if 'url' in df.columns:
  44. df['url'] = "'" + df['url']
  45. df.to_excel(prefix.replace('.xlsx', '')+'.xlsx', index=False)
  46. def db2sheets(self, queries, prefix):
  47. writer = pd.ExcelWriter(prefix.replace('.xlsx', '')+'.xlsx')
  48. for sheet_name, query in queries.items():
  49. df = pd.read_sql_query(query, self.db)
  50. if 'url' in df.columns:
  51. df['url'] = "'" + df['url']
  52. df.to_excel(writer, sheet_name=sheet_name, index=False)
  53. writer.save()
  54. def tb2sheet(self, table):
  55. sql = "SELECT * FROM " + table
  56. self.db2sheet(sql, table)
  57. def insert(self, dt, tb):
  58. for k, v in dt.items():
  59. if isinstance(v, str):
  60. dt[k] = v.replace("'", '').strip()
  61. ls = [(k, v) for k, v in dt.items() if v is not None]
  62. sql = 'INSERT INTO %s (' % tb + ','.join(i[0] for i in ls) + \
  63. ') VALUES (' + ','.join('%r' % i[1] for i in ls) + ')'
  64. self.commit(sql)
  65. def insert_clob(self, dt, tb, clob):
  66. for k, v in dt.items():
  67. if isinstance(v, str):
  68. dt[k] = v.replace("'", '').strip()
  69. # 把超长文本保存在一个变量中
  70. # declare = "DECLARE variate CLOB := '%s';\n" % dt[clob]
  71. join = lambda x: '||'.join("'%s'" % x[10922*i: 10922*(i+1)] for i in range(len(x)//10922+1)) # 32768//3
  72. declare = "DECLARE variate CLOB := %s;\n" % join(dt[clob])
  73. dt[clob] = 'variate'
  74. ls = [(k, v) for k, v in dt.items() if v is not None]
  75. sql = 'INSERT INTO %s (' % tb + ','.join(i[0] for i in ls) + ') VALUES (' +\
  76. ','.join('%r' % i[1] for i in ls) + ');'
  77. sql = declare + 'BEGIN\n%s\nEND;' % sql.replace("'variate'", 'variate')
  78. self.commit(sql)
  79. def update(self, dt_update, dt_condition, table):
  80. sql = 'UPDATE %s SET ' % table + ','.join('%s=%r' % (k, v) for k, v in dt_update.items()) \
  81. + ' WHERE ' + ' AND '.join('%s=%r' % (k, v) for k, v in dt_condition.items())
  82. self.commit(sql)
  83. def truncate(self, tb):
  84. self.commit('truncate table ' + tb)
  85. db_read = Oracle()
  86. fetchall = db_read.fetchall
  87. fetchone = db_read.fetchone
  88. read_clob = db_read.read_clob
  89. if __name__ == '__main__':
  90. query = '''
  91. '''.strip()
  92. for i in fetchone(query, 99):
  93. print(i)

 

conf

 


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

 

文本字符串查询

 


 
  1. class INSTR(Oracle):
  2. """文本字符串查询"""
  3. def highlight_instr(self, table, field, keyword, clob=True):
  4. sql = "SELECT %s FROM %s WHERE INSTR(%s,'%s')>0" % (field, table, field, keyword)
  5. if clob:
  6. for i in self.read_clob(sql):
  7. self.highlight(i, keyword)
  8. else:
  9. for i, in self.fetchone(sql):
  10. self.highlight(i, keyword)
  11. def regexp_instr(self, table, field, pattern, regexp=True, clob=True):
  12. sql = "SELECT %s FROM %s WHERE INSTR(%s,'%s')>0" % (field, table, field, pattern)
  13. sql = sql.replace('INSTR', 'REGEXP_INSTR') if regexp else sql
  14. if clob:
  15. for i in self.read_clob(sql):
  16. yield i
  17. else:
  18. for i, in self.fetchone(sql):
  19. yield i

 

一个简单的建表示例

 


 
  1. -- 建表
  2. CREATE TABLE table_name
  3. (
  4. serial_number NUMBER(10),
  5. collect_date DATE,
  6. url VARCHAR2(255),
  7. long_text CLOB,
  8. price NUMBER(10)-- 若需要精确到小数点2位,按分存储,/100还原到元
  9. );
  10. -- 给表添加备注
  11. COMMENT ON TABLE table_name IS '中文表名';
  12. -- 给表字段添加备注
  13. COMMENT ON COLUMN table_name.serial_number IS '编号';
  14. COMMENT ON COLUMN table_name.collect_date IS '日期';
  15. COMMENT ON COLUMN table_name.url IS 'URL';
  16. COMMENT ON COLUMN table_name.long_text IS '长文本';
  17. COMMENT ON COLUMN table_name.price IS '价钱';
  18. -- 插入
  19. INSERT INTO table_name(collect_date) VALUES (DATE'2019-08-23');
  20. INSERT INTO table_name(long_text) VALUES ('a');
  21. INSERT INTO table_name(long_text) VALUES ('b');
  22. -- 查询
  23. SELECT * FROM table_name WHERE TO_CHAR(long_text) in ('a','b');
  24. -- 查建表语句(表名大写)
  25. SELECT dbms_metadata.get_ddl('TABLE','TABLE_NAME') FROM dual;
  26. -- 删表
  27. DROP TABLE table_name;

 

sqlalchemy

 


 
  1. import os # 解决【UnicodeEncodeError: 'ascii' codec can't encode character】问题
  2. os.environ['NLS_LANG'] = 'AMERICAN_AMERICA.AL32UTF8'
  3. # os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
  4. from cx_Oracle import makedsn
  5. from sqlalchemy import create_engine, Column, String, Integer
  6. from sqlalchemy.ext.declarative import declarative_base
  7. from sqlalchemy.orm import sessionmaker
  8. # 连接数据库(ORA-12505: TNS:listener does not currently know of SID given in connect descriptor)
  9. ip = ''
  10. port = ''
  11. tnsname = '' # 实例名
  12. uname = '' # 用户名
  13. pwd = '' # 密码
  14. dsnStr = makedsn(ip, port, service_name=tnsname)
  15. connect_str = "oracle://%s:%s@%s" % (uname, pwd, dsnStr)
  16. # 创建连接引擎,这个engine是lazy模式,直到第一次被使用才真实创建
  17. engine = create_engine(connect_str, encoding='utf-8')
  18. # 创建对象的基类
  19. Base = declarative_base()
  20. class Student(Base):
  21. # 表名
  22. __tablename__ = 'student'
  23. # 表字段
  24. sid = Column(String(20), primary_key=True)
  25. age = Column(Integer)
  26. # 建表(继承Base的所有表)
  27. Base.metadata.create_all(bind=engine)
  28. # 使用ORM操作数据库
  29. Session = sessionmaker(bind=engine) # 创建ORM基类
  30. session = Session() # 创建ORM对象
  31. tb_obj = Student(sid='a6', age=18) # 创建表对象
  32. session.add(tb_obj) # 添加到ORM对象(插入数据)
  33. session.commit() # 提交
  34. session.close() # 关闭ORM对象
  35. # 删表(继承Base的所有表)
  36. 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之环境配置、实例代码及报错解决方法请查看下面的相关链接

分享到:

相关信息

系统教程栏目

栏目热门教程

人气教程排行

站长推荐

热门系统下载

公众号