一、cx-Oracle 库核心介绍
cx-Oracle 是一款专门用于 Python 程序连接 Oracle 数据库的扩展库，能够实现对 Oracle 数据库的查询、插入、更新、删除等各类操作。其工作原理是基于 Oracle Call Interface（OCI）构建，通过调用 Oracle 客户端的底层接口，建立 Python 与 Oracle 数据库之间的通信桥梁，实现数据的高效交互。
该库的优点十分突出，兼容性强，支持 Python 3.x 系列版本和多种 Oracle 数据库版本，数据传输效率高，能直接处理 Oracle 特有的数据类型；缺点则是安装时需要依赖 Oracle 客户端库，配置步骤相对繁琐，且仅专注于 Oracle 数据库，不支持其他类型数据库。cx-Oracle 采用 BSD 开源许可证，用户可自由下载、使用、修改和分发，无商业授权限制。
二、cx-Oracle 安装与环境配置
2.1 安装前准备
在安装 cx-Oracle 之前，必须先安装 Oracle 客户端库，这是连接 Oracle 数据库的前提条件。Oracle 提供了两种轻量级客户端包供选择：
- Oracle Instant Client：适用于大多数场景，体积小，安装便捷，可从 Oracle 官网下载对应操作系统版本（Windows、Linux、macOS）。
- Oracle Full Client：功能更全面，包含更多开发工具，适合专业数据库开发人员。
以 Windows 系统为例，安装 Oracle Instant Client 的步骤如下：
- 访问 Oracle 官网下载页面（https://www.oracle.com/database/technologies/instant-client/downloads.html），选择与自己系统位数（32 位/64 位）匹配的 Instant Client 包。
- 将下载的压缩包解压到指定目录，例如
D:\oracle\instantclient_21_9。 - 配置系统环境变量：
- 新增环境变量
ORACLE_HOME，值为解压路径D:\oracle\instantclient_21_9。 - 将
D:\oracle\instantclient_21_9添加到系统PATH环境变量中。
- 新增环境变量
2.2 安装 cx-Oracle 库
完成 Oracle 客户端配置后，即可通过 Python 的包管理工具 pip 安装 cx-Oracle，打开命令提示符（CMD）或终端，执行以下命令：
pip install cx-Oracle安装完成后，可在 Python 交互环境中执行 import cx_Oracle 测试是否安装成功，若没有报错，则说明安装完成。
三、cx-Oracle 核心使用方法与代码实例
3.1 建立数据库连接
使用 cx-Oracle 连接 Oracle 数据库，需要提供用户名、密码和数据库连接字符串。连接字符串的格式通常为 主机名/IP地址:端口号/服务名，具体格式需根据数据库配置调整。
代码实例：
import cx_Oracle
# 数据库连接信息
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl" # 本地数据库示例，orcl为服务名
# 建立连接
try:
connection = cx_Oracle.connect(username, password, dsn)
print("数据库连接成功！")
except cx_Oracle.Error as error:
print(f"数据库连接失败：{error}")
finally:
# 关闭连接
if 'connection' in locals() and connection:
connection.close()
print("连接已关闭")代码说明：
- 首先导入 cx_Oracle 库，定义数据库的用户名、密码和连接字符串
dsn。 - 使用
cx_Oracle.connect()方法建立连接，该方法返回一个连接对象。 - 通过
try-except捕获连接过程中可能出现的异常，例如用户名密码错误、网络不通等。 - 最后在
finally块中关闭连接，确保无论连接是否成功，都能释放资源。
3.2 执行基础 SQL 查询
建立数据库连接后，需要创建游标对象来执行 SQL 语句。游标对象是 cx-Oracle 执行 SQL 操作的核心载体，支持查询、插入、更新等操作。
代码实例：查询表数据
import cx_Oracle
# 数据库连接信息
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl"
# 建立连接并创建游标
connection = None
try:
connection = cx_Oracle.connect(username, password, dsn)
cursor = connection.cursor() # 创建游标对象
# 执行查询 SQL 语句
sql = "SELECT empno, ename, job, sal FROM emp WHERE deptno = :deptno"
deptno = 20 # 查询20号部门的员工信息
cursor.execute(sql, deptno=deptno) # 绑定参数，防止SQL注入
# 获取查询结果的两种方式
# 方式1：逐行获取
print("20号部门员工信息（逐行获取）：")
for row in cursor:
empno, ename, job, sal = row
print(f"员工编号：{empno}, 姓名：{ename}, 职位：{job}, 薪资：{sal}")
# 方式2：一次性获取所有结果
cursor.execute(sql, deptno=deptno)
rows = cursor.fetchall() # 获取所有行数据
print("\n20号部门员工信息（一次性获取）：")
for row in rows:
print(f"员工编号：{row[0]}, 姓名：{row[1]}, 职位：{row[2]}, 薪资：{row[3]}")
except cx_Oracle.Error as error:
print(f"执行查询失败：{error}")
finally:
# 关闭游标和连接
if 'cursor' in locals() and cursor:
cursor.close()
if 'connection' in locals() and connection:
connection.close()代码说明：
- 游标对象通过
connection.cursor()创建，后续所有 SQL 操作都通过游标执行。 - 执行查询语句时，使用
cursor.execute(sql, 参数)方法，其中:deptno是参数占位符，通过传入deptno=20绑定参数，这种方式能有效防止 SQL 注入攻击。 - 获取查询结果有两种常用方式：
- 直接遍历游标对象，逐行读取数据。
- 使用
cursor.fetchall()方法一次性获取所有结果，返回一个包含所有行的列表，每行数据是一个元组。
- 操作完成后，需依次关闭游标和连接，释放资源。
3.3 执行数据插入操作
cx-Oracle 支持向 Oracle 数据库插入单条或多条数据，插入操作同样通过游标对象执行，执行后需要调用 connection.commit() 提交事务，否则数据不会真正写入数据库。
代码实例：插入单条数据
import cx_Oracle
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl"
connection = None
try:
connection = cx_Oracle.connect(username, password, dsn)
cursor = connection.cursor()
# 插入数据的 SQL 语句
insert_sql = """
INSERT INTO emp (empno, ename, job, mgr, hiredate, sal, comm, deptno)
VALUES (:empno, :ename, :job, :mgr, :hiredate, :sal, :comm, :deptno)
"""
# 定义要插入的数据
emp_data = {
"empno": 7999,
"ename": "LIU",
"job": "CLERK",
"mgr": 7788,
"hiredate": cx_Oracle.Date(2024, 1, 10), # Oracle日期类型
"sal": 2500,
"comm": None,
"deptno": 20
}
# 执行插入操作
cursor.execute(insert_sql, **emp_data)
connection.commit() # 提交事务
print(f"成功插入 {cursor.rowcount} 条数据")
except cx_Oracle.Error as error:
connection.rollback() # 出错时回滚事务
print(f"插入数据失败：{error}")
finally:
if 'cursor' in locals() and cursor:
cursor.close()
if 'connection' in locals() and connection:
connection.close()代码说明：
- 插入 SQL 语句中使用多个参数占位符（
:empno、:ename等），通过字典emp_data传递参数，使用**解包字典。 - 对于 Oracle 的日期类型，需要使用
cx_Oracle.Date()方法创建对应的日期对象，确保数据类型匹配。 - 执行插入操作后，必须调用
connection.commit()提交事务，否则数据不会持久化到数据库；若出现异常，需调用connection.rollback()回滚事务，避免数据不一致。 cursor.rowcount属性可以获取受影响的行数，用于判断插入操作是否成功。
代码实例：批量插入数据
当需要插入大量数据时，使用 cursor.executemany() 方法可以显著提高效率，该方法支持批量执行 SQL 语句。
import cx_Oracle
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl"
connection = None
try:
connection = cx_Oracle.connect(username, password, dsn)
cursor = connection.cursor()
# 批量插入的 SQL 语句
batch_insert_sql = """
INSERT INTO emp (empno, ename, job, deptno)
VALUES (:empno, :ename, :job, :deptno)
"""
# 批量数据列表
batch_data = [
(8001, "ZHANG", "ANALYST", 20),
(8002, "WANG", "SALESMAN", 30),
(8003, "ZHAO", "MANAGER", 10)
]
# 执行批量插入
cursor.executemany(batch_insert_sql, batch_data)
connection.commit()
print(f"成功批量插入 {cursor.rowcount} 条数据")
except cx_Oracle.Error as error:
connection.rollback()
print(f"批量插入失败：{error}")
finally:
if 'cursor' in locals() and cursor:
cursor.close()
if 'connection' in locals() and connection:
connection.close()代码说明：
cursor.executemany()方法接收两个参数，第一个是 SQL 语句，第二个是包含多条数据的列表，列表中的每个元素是一个元组，对应 SQL 语句中的参数。- 批量插入相比多次执行单条插入，减少了网络交互和数据库事务的开销，效率更高，适合大数据量的插入场景。
3.4 执行数据更新与删除操作
数据的更新和删除操作与插入操作类似，都是通过游标执行 SQL 语句，然后提交事务。
代码实例：更新数据
import cx_Oracle
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl"
connection = None
try:
connection = cx_Oracle.connect(username, password, dsn)
cursor = connection.cursor()
# 更新 SQL 语句：更新7999号员工的薪资
update_sql = "UPDATE emp SET sal = :new_sal WHERE empno = :empno"
new_sal = 3000
empno = 7999
cursor.execute(update_sql, new_sal=new_sal, empno=empno)
connection.commit()
print(f"成功更新 {cursor.rowcount} 条数据")
except cx_Oracle.Error as error:
connection.rollback()
print(f"更新数据失败：{error}")
finally:
if 'cursor' in locals() and cursor:
cursor.close()
if 'connection' in locals() and connection:
connection.close()代码实例：删除数据
import cx_Oracle
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl"
connection = None
try:
connection = cx_Oracle.connect(username, password, dsn)
cursor = connection.cursor()
# 删除 SQL 语句：删除8003号员工
delete_sql = "DELETE FROM emp WHERE empno = :empno"
empno = 8003
cursor.execute(delete_sql, empno=empno)
connection.commit()
print(f"成功删除 {cursor.rowcount} 条数据")
except cx_Oracle.Error as error:
connection.rollback()
print(f"删除数据失败：{error}")
finally:
if 'cursor' in locals() and cursor:
cursor.close()
if 'connection' in locals() and connection:
connection.close()代码说明：
- 更新和删除操作的 SQL 语句同样使用参数占位符，避免 SQL 注入。
- 执行后通过
cursor.rowcount查看受影响的行数，若行数为 0，说明没有符合条件的数据。
3.5 处理 Oracle 特有的数据类型
Oracle 数据库包含一些特有的数据类型，例如 NUMBER、DATE、TIMESTAMP、CLOB 等，cx-Oracle 提供了对应的处理方式。
代码实例：处理 CLOB 大文本类型
import cx_Oracle
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl"
connection = None
try:
connection = cx_Oracle.connect(username, password, dsn)
cursor = connection.cursor()
# 假设存在一个表 test_clob，包含 id 和 content 字段，content 为 CLOB 类型
# 插入 CLOB 数据
insert_clob_sql = "INSERT INTO test_clob (id, content) VALUES (:id, :content)"
clob_content = cx_Oracle.CLOB(connection) # 创建 CLOB 对象
clob_content.write("这是一段很长的文本内容，用于测试 CLOB 数据类型的处理方式。" * 100)
cursor.execute(insert_clob_sql, id=1, content=clob_content)
connection.commit()
print("CLOB 数据插入成功")
# 查询 CLOB 数据
select_clob_sql = "SELECT content FROM test_clob WHERE id = :id"
cursor.execute(select_clob_sql, id=1)
clob_data = cursor.fetchone()[0]
print(f"CLOB 数据内容（前200字）：{clob_data.read()[:200]}")
except cx_Oracle.Error as error:
connection.rollback()
print(f"处理 CLOB 数据失败：{error}")
finally:
if 'cursor' in locals() and cursor:
cursor.close()
if 'connection' in locals() and connection:
connection.close()代码说明：
- 对于 CLOB 类型的数据，需要先通过
cx_Oracle.CLOB(connection)创建 CLOB 对象，然后使用write()方法写入文本内容。 - 查询 CLOB 数据时，获取到的是 CLOB 对象，通过
read()方法可以读取其中的文本内容。
四、cx-Oracle 实际应用案例：员工薪资管理系统
下面结合一个简单的员工薪资管理系统案例，综合展示 cx-Oracle 的使用方法。该案例实现以下功能：
- 连接 Oracle 数据库。
- 查询指定部门的员工薪资信息。
- 给指定员工涨薪。
- 新增员工记录。
完整代码实例：
import cx_Oracle
from datetime import datetime
class EmpSalaryManager:
def __init__(self, username, password, dsn):
"""初始化数据库连接信息"""
self.username = username
self.password = password
self.dsn = dsn
self.connection = None
self.cursor = None
def connect_db(self):
"""建立数据库连接"""
try:
self.connection = cx_Oracle.connect(self.username, self.password, self.dsn)
self.cursor = self.connection.cursor()
print("数据库连接成功！")
except cx_Oracle.Error as error:
print(f"连接失败：{error}")
raise
def disconnect_db(self):
"""关闭数据库连接"""
if self.cursor:
self.cursor.close()
if self.connection:
self.connection.close()
print("数据库连接已关闭")
def query_dept_salary(self, deptno):
"""查询指定部门的员工薪资信息"""
try:
sql = "SELECT empno, ename, sal, hiredate FROM emp WHERE deptno = :deptno"
self.cursor.execute(sql, deptno=deptno)
results = self.cursor.fetchall()
if not results:
print(f"未查询到{deptno}号部门的员工信息")
return
print(f"\n{deptno}号部门员工薪资信息：")
print("-" * 50)
print(f"{'员工编号':<10}{'姓名':<10}{'薪资':<10}{'入职日期':<15}")
print("-" * 50)
for empno, ename, sal, hiredate in results:
# 格式化日期显示
hiredate_str = hiredate.strftime("%Y-%m-%d") if hiredate else "未知"
print(f"{empno:<10}{ename:<10}{sal:<10}{hiredate_str:<15}")
except cx_Oracle.Error as error:
print(f"查询失败：{error}")
def update_salary(self, empno, add_sal):
"""给指定员工涨薪"""
try:
# 先查询原薪资
check_sql = "SELECT sal FROM emp WHERE empno = :empno"
self.cursor.execute(check_sql, empno=empno)
result = self.cursor.fetchone()
if not result:
print(f"未找到编号为{empno}的员工")
return
old_sal = result[0]
new_sal = old_sal + add_sal
# 更新薪资
update_sql = "UPDATE emp SET sal = :new_sal WHERE empno = :empno"
self.cursor.execute(update_sql, new_sal=new_sal, empno=empno)
self.connection.commit()
print(f"\n员工{empno}涨薪成功！原薪资：{old_sal}, 新薪资：{new_sal}")
except cx_Oracle.Error as error:
self.connection.rollback()
print(f"涨薪失败：{error}")
def add_employee(self, emp_data):
"""新增员工记录"""
try:
insert_sql = """
INSERT INTO emp (empno, ename, job, mgr, hiredate, sal, comm, deptno)
VALUES (:empno, :ename, :job, :mgr, :hiredate, :sal, :comm, :deptno)
"""
self.cursor.execute(insert_sql, **emp_data)
self.connection.commit()
print(f"\n新增员工{emp_data['ename']}成功！员工编号：{emp_data['empno']}")
except cx_Oracle.Error as error:
self.connection.rollback()
print(f"新增员工失败：{error}")
# 主程序入口
if __name__ == "__main__":
# 数据库配置
username = "scott"
password = "tiger"
dsn = "127.0.0.1:1521/orcl"
# 创建薪资管理对象
manager = EmpSalaryManager(username, password, dsn)
try:
# 连接数据库
manager.connect_db()
# 1. 查询20号部门的薪资信息
manager.query_dept_salary(20)
# 2. 给7999号员工涨薪500
manager.update_salary(7999, 500)
# 3. 新增员工
new_emp = {
"empno": 8004,
"ename": "CHEN",
"job": "ENGINEER",
"mgr": 7782,
"hiredate": cx_Oracle.Date.today(),
"sal": 4000,
"comm": 0,
"deptno": 20
}
manager.add_employee(new_emp)
# 再次查询20号部门信息，查看新增和更新结果
manager.query_dept_salary(20)
except Exception as e:
print(f"程序执行异常：{e}")
finally:
# 关闭连接
manager.disconnect_db()代码说明：
- 该案例通过面向对象的方式封装了员工薪资管理的功能，
EmpSalaryManager类包含连接数据库、查询薪资、更新薪资、新增员工等方法，代码结构清晰，便于维护和扩展。 - 在
update_salary方法中，先查询员工原薪资，再计算新薪资并更新，确保操作的准确性；在add_employee方法中，使用cx_Oracle.Date.today()获取当前日期作为入职日期。 - 主程序中调用类的方法，依次执行查询、涨薪、新增员工操作，并再次查询验证结果，展示了 cx-Oracle 在实际项目中的综合应用。
五、相关资源地址
- Pypi地址：https://pypi.org/project/cx-Oracle
- Github地址：https://github.com/oracle/python-cx_Oracle
- 官方文档地址：https://cx-oracle.readthedocs.io/en/latest/
关注我，每天分享一个实用的Python自动化工具。
