本文主要是介绍在Oracle DevGym上学习cx_Oracle的笔记，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
本文是在Oracle DevGym上学习cx_Oracle的笔记。
cx_Oracle帮助文档参见这里
更多学习资源参加Oracle在GitHub上的项目。
环境准备
使用Vagrant Box安装的Oracle Database 19c环境，操作系统为Oracle Linux 7。 详见这里
安装Python3
$ sudo yum install python3
$ python3
Python 3.6.8 (default, Aug 7 2019, 08:02:28)
[GCC 4.8.5 20150623 (Red Hat 4.8.5-39.0.1)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>>
安装cx_Oracle
$ pip3 install cx_Oracle --user --upgrade
Collecting cx_OracleDownloading https://files.pythonhosted.org/packages/d5/15/d38862a4bd0e18d8ef2a3c98f39e743b8951ec5efd8bc63e75db04b9bc31/cx_Oracle-7.3.0-cp36-cp36m-manylinux1_x86_64.whl (737kB)100% |████████████████████████████████| 737kB 23kB/s
Installing collected packages: cx-Oracle
Successfully installed cx-Oracle-7.3.0
cx_Oracle依赖于Oracle client library连接Oracle数据库，在我们的Vagrant Box中已经包含了client library。
数据库和cx_Oracle模块就绪后，接下来安装示例Schema，参见如何使用github安装Oracle 数据库12c Sample Schema
测试cx_Oracle
从页面下载示例代码。
只需修改connection = ...那一行，填写相应的hr用户口令和服务名，命名此文件为cxdemo.py。
import cx_Oracle# Establish the database connection
connection = cx_Oracle.connect("hr", "oracle", "localhost/orclpdb1")# Obtain a cursor
cursor = connection.cursor()# Data for binding
managerId = 145
firstName = "Peter"# Execute the query
sql = """SELECT first_name, last_nameFROM employeesWHERE manager_id = :mid AND first_name = :fn"""
cursor.execute(sql, mid = managerId, fn = firstName)# Loop over the result set
for row in cursor:print(row)
运行以上程序输出如下：
('Peter', 'Hall')
('Peter', 'Tucker')
DEV GYM教程Using CX_ORACLE and Python
所有示例代码都可以从这里下载。不过其中有错误，我已在本文中指出。
在这一节的开始，作者提到了两个ORM软件，SQLAlchemy和Pony。
作者引用了Martin Fowler在文章OrmHate中的一句话:
“Mapping to a relational database involves lots of repetitive, boiler-plate code. A framework that allows me to avoid 80% of that is worthwhile even if it is only 80%.”
意思是说，ORM即使只解决了80%的问题，也是值得的。我看了整个文章，后面其实隐含的意思是ORM是做不到100%的，还有一部分需要充分了解数据库，或直接对数据库进行CRUD操作，类似于cx_Oracle这样。
准备工作
这一部分的教程参见这里。
这一部分设置了教程使用的测试表结构数据，我们仍安装在hr用户下。 注意两个表的id列定义，
CREATE TABLE cx_people (id NUMBER GENERATED BY DEFAULT AS identity,name VARCHAR2(20),age NUMBER,notes VARCHAR2(100)
)
/ALTER TABLE CX_PEOPLE
ADD CONSTRAINT PK_CX_PEOPLE PRIMARY KEY ("ID")
/CREATE TABLE CX_PETS (id NUMBER GENERATED BY DEFAULT AS IDENTITY,name VARCHAR2(20),owner NUMBER,type VARCHAR2(100)
)
/ALTER TABLE CX_PETS ADD CONSTRAINT PK_CX_PETS PRIMARY KEY ("ID")
/ALTER TABLE CX_PETS ADD CONSTRAINT FK_CX_PETS_OWNER FOREIGN KEY ("OWNER") REFERENCES "CX_PEOPLE" ("ID")
/INSERT INTO cx_people (name, age, notes)VALUES ('Bob', 35, 'I like dogs')
/INSERT INTO cx_people (name, age, notes)VALUES ('Kim', 27, 'I like birds')
/INSERT INTO cx_pets (name, owner, type)VALUES ('Duke', 1, 'dog')
/INSERT INTO cx_pets (name, owner, type)VALUES ('Pepe', 2, 'bird')
/COMMIT
/
输出如下：
SQL> desc cx_peopleName Null? Type----------------------------------------- -------- ----------------------------ID NOT NULL NUMBERNAME VARCHAR2(20)AGE NUMBERNOTES VARCHAR2(100)SQL> desc cx_petsName Null? Type----------------------------------------- -------- ----------------------------ID NOT NULL NUMBERNAME VARCHAR2(20)OWNER NUMBERTYPE VARCHAR2(100)SQL> select * from cx_people;ID NAME AGE NOTES
---------- -------- ---------- ------------------------1 Bob 35 I like dogs2 Kim 27 I like birdsSQL> select * from cx_pets;ID NAME OWNER TYPE
---------- -------- ---------- ------------1 Duke 1 dog2 Pepe 2 bird
如需清理测试数据，执行以下:
drop table CX_PETS
/drop table CX_PEOPLE
/
注意上面两个表id列的定义，使用了12.1版本后支持的IDENTITY Columns特性，简称为ID列。
简单介绍见这里，记住以下几点：
- 每个表只能有一个ID列
- 当表中数据被清除时，ID列的值重置会初始值
参照以下的实验理解其概念:
create table test(id NUMBER GENERATED BY DEFAULT AS identity, name varchar2(20));
alter table test modify id generated BY DEFAULT as identity (START WITH 3);
insert into test(name) values('tony');
select * from test;ID NAME
---------- --------------------3 tony
连接测试:
>>> import cx_Oracle
>>> con = cx_Oracle.connect('hr/oracle@orclpdb1')
<cx_Oracle.Connection to hr@orclpdb1>
插入数据
这一部分的教程参见这里。
第一个程序reset_data.py用于重置数据，也就是删除所有数据后再插入测试数据，也就是建立测试数据的基线。
import cx_Oracle
import os
connectString = os.getenv('db_connect') # 从环境变量db_connect中读取连接串
con = cx_Oracle.connect(connectString)
cur = con.cursor()# Delete rows
statement = 'delete from cx_pets'
cur.execute(statement)# 设置id列，如果是自动生成(也就是当未指定此列值时)，则从3开始。 因为前面已经插入两行了
statement = 'alter table cx_pets modify id generated BY DEFAULT as identity (START WITH 3)'
cur.execute(statement)# Delete rows
statement = 'delete from cx_people'
cur.execute(statement)# 设置id列，如果是自动生成(也就是当未指定此列值时)，则从3开始。 因为前面已经插入两行了
statement = 'alter table cx_people modify id generated BY DEFAULT as identity (START WITH 3)'
cur.execute(statement)# Insert default rows
rows = [(1, 'Bob', 35, 'I like dogs'), (2, 'Kim', 27, 'I like birds')]
cur.bindarraysize = len(rows)
cur.setinputsizes(int, 20, int, 100) # 20 和 100 表示VARCHAR2的宽度
cur.executemany("insert into cx_people(id, name, age, notes) values (:1, :2, :3, :4)", rows)
con.commit()# Insert default rows
rows = [(1, 'Duke', 1, 'dog'), (2, 'Pepe', 2, 'bird')]
cur.bindarraysize = len(rows)
cur.setinputsizes(int, 20, int, 100) # 20 和 100 表示VARCHAR2的宽度
cur.executemany("insert into cx_pets (id, name, owner, type) values (:1, :2, :3, :4)", rows)
con.commit()cur.close()
测试此程序如下：
$ export db_connect='hr/oracle@orclpdb1'
$ python3 reset_data.py
$ sqlplus登入数据库查看表和数据均存在
接下来是一个程序模板，我们命名为template.py：
import cx_Oracle
import os
connectString = os.getenv('db_connect')
con = cx_Oracle.connect(connectString)def get_all_rows(label):# Query all rowscur = con.cursor()statement = 'select id, name, age, notes from cx_people order by id'cur.execute(statement)res = cur.fetchall()print(label + ': ')print (res)print(' ')cur.close()get_all_rows('Original Data')# 插入你的代码get_all_rows('New Data')
这个模板的作用是显示操作前后表cx_people的所有数据。 之后我们只需将操作代码放置到# 插入你的代码这一行。
基本INSERT操作
在模板中嵌入以下代码并形成文件insert.py：
cur = con.cursor()
statement = 'insert into cx_people(name, age, notes) values (:2, :3, :4)'
cur.execute(statement, ('Sandy', 31, 'I like horses'))
con.commit()
运行结果为:
Original Data:
[(1, 'Bob', 35, 'I like dogs'), (2, 'Kim', 27, 'I like birds')]New Data:
[(1, 'Bob', 35, 'I like dogs'), (2, 'Kim', 27, 'I like birds'), (3, 'Sandy', 31, 'I like horses')]
注意插入时并没有指定id的值，这一列是自动增长的，且初始值设为3。
如果需要插入多行，可以执行多个execute()语句，然后一次性用commit()提交。
利用cx_Oracle.connect()可以在一个程序里创建多个数据库连接，也就是可以模拟多个会话，例如作者创建了两个连接来验证只有在第一个会话提交数据后，第二个会话才能看到数据的变化。
利用RETURNING返回参数
重置数据，然后将以下代码插入到模板并运行, 原程序有错误，我已经做了修改，见中文注释：
cur = con.cursor()new_id = cur.var(cx_Oracle.NUMBER)statement = 'insert into cx_people(name, age, notes) values (:1, :2, :3) returning id into :4'
cur.execute(statement, ('Sandy', 31, 'I like horses', new_id))sandy_id = new_id.getvalue()[0] # getvalue()返回List，因此我们只取其第一个元素pet_statement = 'insert into cx_pets (name, owner, type) values (:1, :2, :3)'
cur.execute(pet_statement, ('Big Red', sandy_id, 'horse'))
con.commit()print('Our new value is: ' + str(sandy_id).rstrip('.0'))sandy_pet_statement = 'select name, owner, type from cx_pets where owner = :owner'
cur.execute(sandy_pet_statement, {'owner': int(sandy_id)}) # 转换成整形以避免类型不匹配错误
res = cur.fetchall()
print('Sandy\'s pets: ')
print (res)
print(' ')
上面代码中最重要的是returning关键字，使得我们可以在SQL中输出变量。
原代码中的sandy_id = new_id.getvalue()必须修改，否则报以下错误：
cx_Oracle.DatabaseError: ORA-01484: arrays can only be bound to PL/SQL statements
下面一段代码和上一个类似，只不过多返回了一个变量，注意原代码同样存在错误:
cur = con.cursor()new_id = cur.var(cx_Oracle.NUMBER)
new_name = cur.var(cx_Oracle.STRING)statement = 'insert into cx_people(name, age, notes) values (:1, :2, :3) returning id, name into :4, :5'
cur.execute(statement, ('Sandy', 31, 'I like horses', new_id, new_name))sandy_id = new_id.getvalue()[0] # 原代码是错的
sandy_name = new_name.getvalue()[0] # 原代码是错的con.commit()print('Our new id is: ' + str(sandy_id).rstrip('.0') + ' name: ' + str(sandy_name))
利用EXECUTEMANY()一次性插入多行
每次插入一行效率很低，我们可以利用bind array一次插入多行，使用的方法是executemany()：
rows = [('Sandy', 31, 'I like horses'), ('Suzy', 29, 'I like rabbits')]
cur = con.cursor()
cur.bindarraysize = len(rows)
cur.setinputsizes(20, int, 100) # 20和100是列定义VARCHAR2的长度
statement = 'insert into cx_people(name, age, notes) values (:1, :2, :3)'
cur.executemany(statement, rows)
con.commit()
查询数据
这一部分的教程参见这里。
基本查询
以下的代码返回表cx_people中的数据:
cur = con.cursor()
statement = 'select id, name, age, notes from cx_people'
cur.execute(statement)
res = cur.fetchall()
print (res)
输出为list of tuple：
[(1, 'Bob', 35, 'I like dogs'), (2, 'Kim', 27, 'I like birds')]
特定查询
当我们需要查询满足特定条件的行时，使用字符串拼接是不建议的，可能带来安全风险，如SQL Injection：
-- 不建议
statement = "select id, name, age, notes from cx_people where name= '" + person_name + "'"
我们建议的方法是使用绑定变量，按位置或按名称：
-- 按位置
cur.execute('select id, name, age, notes from cx_people where name=:1 and age=:2', ('Bob', 35))
-- 按名称
cur.execute('select id, name, age, notes from cx_people where name = :name and age = :age', {'name':'Bob', 'age':35})
代码如下，当绑定变量为按名称时，带入变量使用dictionary的键值对形式：
cur = con.cursor()
person_name = 'Kim'
statement = 'select id, name, age, notes from cx_people where name = :name'
cur.execute(statement, {'name':person_name})
res = cur.fetchall()
print (res)
输出如下：
[(2, 'Kim', 27, 'I like birds')]
更新数据
这一部分的教程参见这里。
在这一节，我们使用了新的模板，变化是函数get_all_rows增加了data_type参数以返回指定表的所有数据：
import cx_Oracle
import os
connectString = os.getenv('db_connect')
con = cx_Oracle.connect(connectString)def get_all_rows(label, data_type='people'):# Query all rowscur = con.cursor()if (data_type == 'pets'):statement = 'select id, name, owner, type from cx_pets order by owner, id'else:statement = 'select id, name, age, notes from cx_people order by id'cur.execute(statement)res = cur.fetchall()print(label + ': ')print (res)print(' ')cur.close()get_all_rows('Original Data')# Your code hereget_all_rows('New Data')
相应的，reset_data.py也修改为如下，唯一的变化是cx_pets表中插入了7条而不是2条数据：
import cx_Oracle
import os
connectString = os.getenv('db_connect')
con = cx_Oracle.connect(connectString)
cur = con.cursor()# Delete rows
statement = 'delete from cx_pets'
cur.execute(statement)# Reset Identity Coulmn
statement = 'alter table cx_pets modify id generated BY DEFAULT as identity (START WITH 8)'
cur.execute(statement)# Delete rows
statement = 'delete from cx_people'
cur.execute(statement)# Reset Identity Coulmn
statement = 'alter table cx_people modify id generated BY DEFAULT as identity (START WITH 3)'
cur.execute(statement)# Insert default rows
rows = [(1, 'Bob', 35, 'I like dogs'), (2, 'Kim', 27, 'I like birds')]
cur.bindarraysize = 2
cur.setinputsizes(int, 20, int, 100)
cur.executemany("insert into cx_people(id, name, age, notes) values (:1, :2, :3, :4)", rows)
con.commit()# Insert default rows
rows = [(1, 'Duke', 1, 'dog'),(2, 'Pepe', 2, 'bird'),(3, 'Princess', 1, 'snake'),(4, 'Polly', 1, 'bird'),(5, 'Rollo', 1, 'horse'),(6, 'Buster', 2, 'dog'),(7, 'Fido', 1, 'cat')]
cur.bindarraysize = len(rows) # 原代码中设置成2
cur.setinputsizes(int, 20, int, 100)
cur.executemany("insert into cx_pets (id, name, owner, type) values (:1, :2, :3, :4)", rows)
con.commit()cur.close()
此代码中cursor.bindarraysize表示每次绑定的行数，如果是1，则每次插入一行。当批量插入时，将此值设大有利于性能。
简单更新
代码片段如下：
cur = con.cursor()
statement = 'update cx_people set age = :1 where id = :2'
cur.execute(statement, (31, 1))
con.commit()
输出为
Original Data:
[(1, 'Bob', 35, 'I like dogs'), (2, 'Kim', 27, 'I like birds')]New Data:
[(1, 'Bob', 31, 'I like dogs'), (2, 'Kim', 27, 'I like birds')]
通过rowcount()获取受影响的行数
由于需要更新cx_pets，因此需要将模板中的get_all_rows()函数修改为以下：
get_all_rows('Original Data', 'pets')# Your code hereget_all_rows('New Data', 'pets')
代码片段如下：
cur = con.cursor()
statement = 'update cx_pets set owner = :1 where id = :2'
cur.execute(statement, (1, 6))
print('Number of rows updated: ' + str(cur.rowcount))
con.commit()
print(' ')
执行结果如下，注意id为6那行，owner由2改为了1：
Original Data:
[(1, 'Duke', 1, 'dog'), (3, 'Princess', 1, 'snake'), (4, 'Polly', 1, 'bird'), (5, 'Rollo', 1, 'horse'), (7, 'Fido', 1, 'cat'), (2, 'Pepe', 2, 'bird'), (6, 'Buster', 2, 'dog')]Number of rows updated: 1New Data:
[(1, 'Duke', 1, 'dog'), (3, 'Princess', 1, 'snake'), (4, 'Polly', 1, 'bird'), (5, 'Rollo', 1, 'horse'), (6, 'Buster', 1, 'dog'), (7, 'Fido', 1, 'cat'), (2, 'Pepe', 2, 'bird')]可以尝试，如果rowcount不满足某一条件，则回退事务。
删除数据
这一部分的教程参见这里。
在这一节，模板函数和上一节一致。
重置文件reset_data.py修改为以下，变化为cx_people和cx_pets都插入了7行。
import cx_Oracle
import os
connectString = os.getenv('db_connect')
con = cx_Oracle.connect(connectString)
cur = con.cursor()# Delete rows
statement = 'delete from cx_pets'
cur.execute(statement)# Reset Identity Coulmn
statement = 'alter table cx_pets modify id generated BY DEFAULT as identity (START WITH 8)'
cur.execute(statement)# Delete rows
statement = 'delete from cx_people'
cur.execute(statement)# Reset Identity Coulmn
statement = 'alter table cx_people modify id generated BY DEFAULT as identity (START WITH 8)'
cur.execute(statement)# Insert default rows
rows = [(1, 'Bob', 35, 'I like dogs'),(2, 'Kim', 27, 'I like birds'),(3, 'Cheryl', 23, 'I like horses'),(4, 'Bob', 27, 'I like rabbits'),(5, 'Stacey', 45, 'I like snakes'),(6, 'Pete', 23, 'I like cats'),(7, 'Pat', 36, 'I like dogs')]
cur.bindarraysize = 2
cur.setinputsizes(int, 20, int, 100)
cur.executemany("insert into cx_people(id, name, age, notes) values (:1, :2, :3, :4)", rows)
con.commit()# Insert default rows
rows = [(1, 'Duke', 1, 'dog'),(2, 'Dragon', 2, 'bird'),(3, 'Sneaky', 5, 'snake'),(4, 'Red', 2, 'bird'),(5, 'Red', 3, 'horse'),(6, 'Buster', 2, 'dog'),(7, 'Fido', 7, 'cat')]
cur.bindarraysize = 2
cur.setinputsizes(int, 20, int, 100)
cur.executemany("insert into cx_pets (id, name, owner, type) values (:1, :2, :3, :4)", rows)
con.commit()cur.close()
代码片段如下：
cur = con.cursor()
statement = 'delete from cx_pets where id = :id'
cur.execute(statement, {'id':1})
con.commit()
由于都和前面差不多，因此不再赘述。
遗留问题：如果绑定Null值?
cx_Oracle Connection Object
参见文档
连接数据库成功后会返回cx_Oracle.Connection对象。
Connection对象的方法和属性包括:
>>> connection.
connection.DataError( connection.cancel( connection.encoding connection.outputtypehandler
connection.DatabaseError( connection.changepassword( connection.enq( connection.ping(
connection.Error( connection.client_identifier connection.enqoptions( connection.prepare(
connection.IntegrityError( connection.clientinfo connection.external_name connection.queue(
connection.InterfaceError( connection.close( connection.getSodaDatabase( connection.rollback(
connection.InternalError( connection.commit( connection.gettype( connection.shutdown(
connection.NotSupportedError( connection.createlob( connection.handle connection.startup(
connection.OperationalError( connection.current_schema connection.inputtypehandler connection.stmtcachesize
connection.ProgrammingError( connection.cursor( connection.internal_name connection.subscribe(
connection.Warning( connection.dbop connection.ltxid connection.tag
connection.action connection.deq( connection.maxBytesPerCharacter connection.tnsentry
connection.autocommit connection.deqoptions( connection.module connection.unsubscribe(
connection.begin( connection.dsn connection.msgproperties( connection.username
connection.callTimeout connection.edition connection.nencoding connection.version
除了建立和关闭连接，此对象还可切换schema，关闭和启动数据库。
以下是连接成功后其它一些属性：
>>> connection.dsn
'localhost/orclpdb1'
>>> connection.autocommit
0
>>> connection.encoding
'ASCII'
>>> connection.version
'19.3.0.0.0'
>>> connection.username
'hr'
>>> connection.tnsentry
'localhost/orclpdb1'
>>> connection.nencoding
'ASCII'
>>> connection.current_schema
>>> connection.maxBytesPerCharacter
1
Connect to the database using cx_Oracle
https://devgym.oracle.com/pls/apex/f?p=10001:5:112176356535023::NO:5:P5_COMP_EVENT_ID,P5_QUIZ_ID,P5_CLASS_ID,P5_WORKOUT_ID,P5_RETURN_TO_PAGE,P5_QUESTION_ID,P5_PREVIEW_ONLY,P5_USER_WORKOUT_ID,P5_COMPEV_ANSWER_ID:2015492,3012272,4702,123001,329,N,545944,&cs=1ulwhvsWu95YVZeTClXQNf9g7iu_a5pvJQnnrVTTyxQzOoIOIpZwUXfeHonWMQTwuUiGArC7qBHYoMu1d_KB4IQ
cx_Oracle Cursor Object
Cursor对象是操作数据库最重要的对象。详见这里
Cursor对象的方法和属性如下：
>>> cursor.
cursor.arraysize cursor.close( cursor.fetchmany( cursor.inputtypehandler cursor.scroll(
cursor.arrayvar( cursor.connection cursor.fetchone( cursor.lastrowid cursor.scrollable
cursor.bindarraysize cursor.description cursor.fetchraw( cursor.outputtypehandler cursor.setinputsizes(
cursor.bindnames( cursor.execute( cursor.fetchvars cursor.parse( cursor.setoutputsize(
cursor.bindvars cursor.executemany( cursor.getarraydmlrowcounts( cursor.prepare( cursor.statement
cursor.callfunc( cursor.executemanyprepared( cursor.getbatcherrors( cursor.rowcount cursor.var(
cursor.callproc( cursor.fetchall( cursor.getimplicitresults( cursor.rowfactory
以下为运行cx_demo.py时输出的部分属性：
>>> cursor.arraysize
100
>>> cursor.bindarraysize
1
>>> cursor.bindvars
{'mid': <cx_Oracle.NUMBER with value 145>, 'fn': <cx_Oracle.STRING with value 'Peter'>}
>>> cursor.connection
<cx_Oracle.Connection to hr@localhost/orclpdb1>
>>> cursor.description
[('FIRST_NAME', <class 'cx_Oracle.STRING'>, 20, 20, None, None, 1), ('LAST_NAME', <class 'cx_Oracle.STRING'>, 25, 25, None, None, 0)]
>>> cursor.lastrowid
>>> cursor.rowcount
0
>>> cursor.rowfactory
>>> cursor.scrollable
False
>>> cursor.statement
'SELECT first_name, last_name\n FROM employees\n WHERE manager_id = :mid AND first_name = :fn'
cx_Oracle cursor functionality
https://devgym.oracle.com/pls/apex/f?p=10001:5:112176356535023::NO:5:P5_COMP_EVENT_ID,P5_QUIZ_ID,P5_CLASS_ID,P5_WORKOUT_ID,P5_RETURN_TO_PAGE,P5_QUESTION_ID,P5_PREVIEW_ONLY,P5_USER_WORKOUT_ID,P5_COMPEV_ANSWER_ID:2015492,3012274,4702,123001,329,N,545944,&cs=1Jszzdg_Drxs1fhIP-KKyL0-PdoesilD8yCC2xMU25NdmsrcNoTm1uEwlRJBXPH_jVgJh5jazTIhLjPjdUw_k4w
这篇关于在Oracle DevGym上学习cx_Oracle的笔记的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
