Python实用工具：cx-Oracle 零基础入门教程

作者：

在

一、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自动化工具。

实用工具

Python实用工具：cx-Oracle 零基础入门教程

一、cx-Oracle 库核心介绍

二、cx-Oracle 安装与环境配置

2.1 安装前准备

2.2 安装 cx-Oracle 库

三、cx-Oracle 核心使用方法与代码实例

3.1 建立数据库连接

3.2 执行基础 SQL 查询

3.3 执行数据插入操作

3.4 执行数据更新与删除操作

3.5 处理 Oracle 特有的数据类型

四、cx-Oracle 实际应用案例：员工薪资管理系统

五、相关资源地址

更多文章

Python实用工具Upgini：零代码特征工程与数据集增强教程

Python实用工具库excalibur：PDF表格提取与数据处理实战教程

Python实用工具：rows库快速入门与实战指南

Python实用工具：Camelot库——轻松提取PDF表格数据的完整指南