Python如何高效查询Oracle数据库？-杰瑞科技汇

环境准备
- 安装 cx_Oracle 库
- 安装 Oracle 客户端
建立数据库连接
执行基本查询
参数化查询 (防止 SQL 注入)
处理查询结果
处理大数据集 (使用游标)
执行 DML 语句 (增删改)
使用连接池 (提升性能)
完整代码示例
常见问题与解决方案

环境准备

a. 安装 `cx_Oracle` 库

使用 pip 进行安装,非常简单：

（图片来源网络，侵删）

pip install cx_Oracle

b. 安装 Oracle 客户端

cx_Oracle 是一个 Python 接口，它本身不包含 Oracle 数据库的客户端库，它需要一个本地的 Oracle 客户端来与数据库服务器通信。

有两种主要方式来满足这个依赖：

安装完整的 Oracle 客户端 (Instant Client) 这是最推荐、最灵活的方式,特别是对于部署在服务器上的应用。

下载 Instant Client:
（图片来源网络，侵删）
- 访问 Oracle 官方 Instant Client 下载页面：Oracle Instant Client Downloads
- 根据你的操作系统（如 Linux x86-64, Windows x64, macOS x86-64）和位数选择合适的版本。
- 通常下载 "Basic" 或 "Basic Light" 版本即可。
配置环境变量:
- Windows: 将 Instant Client 的解压路径（C:\oracle\instantclient_19_10）添加到系统的 PATH 环境变量中。
- Linux / macOS: 将解压后的路径添加到 LD_LIBRARY_PATH (Linux) 或 DYLD_LIBRARY_PATH (macOS) 环境变量中。
```
# Linux 示例
export LD_LIBRARY_PATH=/path/to/instantclient_19_10:$LD_LIBRARY_PATH
```
  为了永久生效，可以将此行添加到你的 ~/.bashrc 或 ~/.profile 文件中。

使用 Oracle 官方提供的 cx_Oracle 预编译包 如果你已经安装了某个版本的 Oracle 数据库（在本地开发机上安装了 OracleXE），或者使用 Oracle 提供的特定软件，可能会自动包含所需的库。cx_Oracle 有时会尝试自动定位这些库。

如何验证客户端是否可用？ 安装配置好后，可以在 Python 中尝试导入 cx_Oracle，如果没有报错,通常意味着客户端库可以被找到。

import cx_Oracle
print(cx_Oracle.clientversion())

如果成功输出版本号,说明环境配置正确。

建立数据库连接

连接数据库需要三个基本要素：用户名、密码和连接字符串。

import cx_Oracle
# --- 1. 配置连接信息 ---
# 用户名
username = "your_username"
# 密码
password = "your_password"
# 连接字符串
# 格式: hostname:port/service_name
#  "localhost:1521/XE" 或 "mydb.example.com:1521/ORCLCDB"
dsn = "localhost:1521/XE"
# --- 2. 建立连接 ---
try:
    # 使用 with 语句可以确保连接在使用后被自动关闭
    with cx_Oracle.connect(username, password, dsn) as connection:
        print("成功连接到 Oracle 数据库!")
        # 连接成功后，可以在这里执行操作
        # ...
except cx_Oracle.DatabaseError as e:
    # 捕获数据库错误并打印
    error, = e.args
    print(f"数据库错误代码: {error.code}")
    print(f"数据库错误消息: {error.message}")

执行基本查询

连接成功后，我们需要一个游标来执行 SQL 语句并获取结果。

import cx_Oracle
# ... (连接代码同上) ...
with cx_Oracle.connect(username, password, dsn) as connection:
    print("成功连接到 Oracle 数据库!")
    # 使用 with 语句管理游标
    with connection.cursor() as cursor:
        # --- 3. 执行查询 ---
        sql_query = "SELECT employee_id, first_name, last_name FROM employees WHERE rownum <= 5"
        cursor.execute(sql_query)
        # --- 4. 获取结果 ---
        # fetchone(): 获取下一行结果
        # print(cursor.fetchone())
        # fetchall(): 获取所有剩余的行
        rows = cursor.fetchall()
        print("\n查询结果:")
        for row in rows:
            # row 是一个元组，(101, 'John', 'Doe')
            print(f"ID: {row[0]}, 姓名: {row[1]} {row[2]}")

参数化查询 (防止 SQL 注入)

永远不要使用 Python 的字符串拼接来构建 SQL 语句！ 这会严重导致 SQL 注入漏洞。cx_Oracle 提供了安全的参数化查询方式。

cx_Oracle 使用命名占位符 variable_name。

import cx_Oracle
# ... (连接代码同上) ...
with cx_Oracle.connect(username, password, dsn) as connection:
    with connection.cursor() as cursor:
        # 定义要查询的员工ID
        employee_id_to_find = 101
        # SQL 语句中使用命名占位符
        sql_query = "SELECT first_name, last_name, salary FROM employees WHERE employee_id = :id"
        # execute 方法的第二个参数是一个字典，将占位符名与变量值绑定
        cursor.execute(sql_query, {'id': employee_id_to_find})
        # 获取单行结果
        employee = cursor.fetchone()
        if employee:
            print(f"\n找到员工: {employee[0]} {employee[1]}, 薪资: {employee[2]}")
        else:
            print(f"\n未找到ID为 {employee_id_to_find} 的员工。")

处理查询结果

除了 fetchone() 和 fetchall(),还有其他有用的方法：

fetchmany(size=N): 获取指定数量的行。
rowcount: 属性，表示受 DML 语句影响的行数，或 SELECT 语句返回的行数（在 fetchall() 之后）。
description: 属性，返回一个包含列信息的元组,对于动态处理结果非常有用。

# ... (连接代码同上) ...
with cx_Oracle.connect(username, password, dsn) as connection:
    with connection.cursor() as cursor:
        cursor.execute("SELECT first_name, last_name FROM employees")
        # 一次获取2行
        while True:
            rows = cursor.fetchmany(2)
            if not rows:
                break
            for row in rows:
                print(f"姓名: {row[0]} {row[1]}")

处理大数据集 (使用游标)

当查询返回数百万行数据时，使用 fetchall() 会一次性将所有数据加载到内存中，可能导致内存溢出，正确的做法是使用服务器端游标,逐行或分批处理数据。

默认情况下，cx_Oracle 的游标就是服务器端游标,非常适合处理大数据。

import cx_Oracle
# ... (连接代码同上) ...
with cx_Oracle.connect(username, password, dsn) as connection:
    # 创建一个游标
    cursor = connection.cursor()
    try:
        # 执行一个可能返回大量数据的查询
        cursor.arraysize = 1000  # 可以优化内部获取缓冲区大小
        cursor.execute("SELECT * FROM very_large_table")
        # 逐行处理，内存占用非常低
        print("开始处理大数据集...")
        for i, row in enumerate(cursor):
            if i % 10000 == 0:
                print(f"已处理 {i} 行...")
            # 在这里处理每一行数据 'row'
            # ...
        print("数据处理完成。")
    finally:
        # 确保游标被关闭
        cursor.close()

执行 DML 语句 (增删改)

执行 INSERT, UPDATE, DELETE 语句后，必须调用 connection.commit() 来提交事务，使更改永久生效，默认情况下，cx_Oracle 是在自动提交模式关闭的情况下运行的。

import cx_Oracle
# ... (连接代码同上) ...
with cx_Oracle.connect(username, password, dsn) as connection:
    with connection.cursor() as cursor:
        try:
            # --- INSERT 示例 ---
            new_employee = {
                'id': 999,
                'first_name': 'Python',
                'last_name': 'User',
                'email': 'python.user@example.com',
                'job_id': 'IT_PROG',
                'salary': 9000
            }
            insert_sql = """
                INSERT INTO employees (
                    employee_id, first_name, last_name, email, job_id, salary
                ) VALUES (
                    :id, :first_name, :last_name, :email, :job_id, :salary
                )
            """
            cursor.execute(insert_sql, new_employee)
            print(f"成功插入 {cursor.rowcount} 行新员工。")
            # --- UPDATE 示例 ---
            raise_salary = 1000
            update_sql = "UPDATE employees SET salary = salary + :raise WHERE employee_id = :id"
            cursor.execute(update_sql, {'raise': raise_salary, 'id': 999})
            print(f"成功更新 {cursor.rowcount} 行员工薪资。")
            # --- 提交事务 ---
            connection.commit()
            print("事务已提交。")
        except cx_Oracle.DatabaseError as e:
            # 发生错误时回滚事务
            connection.rollback()
            print("发生错误，事务已回滚。")
            error, = e.args
            print(f"数据库错误: {error.message}")

使用连接池 (提升性能)

对于 Web 应用或需要频繁连接/断开数据库的场景，创建和销毁连接的开销很大，连接池可以预先创建一组数据库连接，供应用程序重复使用,从而显著提高性能。

import cx_Oracle
# --- 1. 创建连接池 ---
# 用户名, 密码, DSN
pool = cx_Oracle.SessionPool(
    user=username,
    password=password,
    dsn=dsn,
    min=2,      # 池中最小连接数
    max=5,      # 池中最大连接数
    increment=1 # 当需要时，每次增加的连接数
)
print("连接池创建成功。")
try:
    # --- 2. 从池中获取连接 ---
    with pool.acquire() as connection:
        print("从连接池中获取了一个连接。")
        with connection.cursor() as cursor:
            cursor.execute("SELECT SYSDATE FROM DUAL")
            sysdate = cursor.fetchone()[0]
            print(f"数据库服务器时间: {sysdate}")
finally:
    # --- 3. 关闭连接池 ---
    pool.close()
    print("连接池已关闭。")

完整代码示例

这是一个综合了连接、查询、参数化、错误处理的完整脚本。

import cx_Oracle
import getpass # 用于安全地输入密码
def main():
    # --- 1. 获取用户输入 ---
    username = input("请输入用户名: ")
    password = getpass.getpass("请输入密码: ") # 密码不会在终端显示
    dsn = input("请输入连接字符串 ( localhost:1521/XE): ")
    # --- 2. 建立连接并执行操作 ---
    try:
        # 使用连接池
        pool = cx_Oracle.SessionPool(user=username, password=password, dsn=dsn, min=1, max=2)
        with pool.acquire() as connection:
            print(f"\n成功连接到 Oracle 数据库 {dsn}!")
            # 示例1: 基本查询
            print("\n--- 示例1: 查询前5名员工 ---")
            with connection.cursor() as cursor:
                cursor.execute("SELECT employee_id, first_name, last_name FROM employees WHERE rownum <= 5")
                for row in cursor:
                    print(f"ID: {row[0]}, 姓名: {row[1]} {row[2]}")
            # 示例2: 参数化查询
            print("\n--- 示例2: 查询特定员工 ---")
            emp_id = input("\n请输入要查询的员工ID: ")
            with connection.cursor() as cursor:
                sql = "SELECT first_name, last_name, salary FROM employees WHERE employee_id = :id"
                cursor.execute(sql, {'id': int(emp_id)})
                employee = cursor.fetchone()
                if employee:
                    print(f"找到员工: {employee[0]} {employee[1]}, 薪资: {employee[2]}")
                else:
                    print(f"未找到ID为 {emp_id} 的员工。")
        # 关闭连接池
        pool.close()
        print("\n操作完成，连接池已关闭。")
    except cx_Oracle.DatabaseError as e:
        error, = e.args
        print(f"\n数据库错误发生:")
        print(f"代码: {error.code}")
        print(f"消息: {error.message}")
    except ValueError:
        print("\n错误: 请输入有效的员工ID。")
    except Exception as e:
        print(f"\n发生未知错误: {e}")
if __name__ == "__main__":
    main()

常见问题与解决方案

DPI-1047: Cannot locate a 64-bit Oracle Client library
- 原因: 这是最常见的错误。cx_Oracle 找不到 Oracle 客户端库。
- 解决方案:
  - 确保你安装的是 64位 的 Python 和 64位 的 Oracle Instant Client。
  - 确保你已经正确配置了系统的 PATH (Windows) 或 LD_LIBRARY_PATH (Linux/macOS) 环境变量，指向 Instant Client 的解压目录。
  - 重启你的终端或 IDE,让环境变量生效。
ORA-12154: TNS:could not resolve the connect identifier specified
- 原因: 连接字符串 dsn 中的服务名 service_name 或 SID sid 不正确，或者 TNS 命名配置文件 (tnsnames.ora) 未找到或配置错误。
- 解决方案:
  - 检查 dsn 字符串是否正确，可以尝试使用 Easy Connect Plus 语法，它更简单，通常不需要 tnsnames.ora 文件，格式如："hostname:port/service_name"。
  - 如果你使用的是 tnsnames.ora 文件，确保该文件位于正确的位置（$ORACLE_HOME/network/admin 或用户家目录下的 .tnsnames.ora）,并且内容正确。
ORA-01017: invalid username/password; logon denied
- 原因: 用户名或密码错误。
- 解决方案: 仔细检查用户名和密码是否正确,注意大小写。

希望这份详细的指南能帮助你在 Python 中顺利地使用 Oracle 数据库！

Python如何高效查询Oracle数据库？

目录

环境准备

a. 安装 `cx_Oracle` 库

b. 安装 Oracle 客户端

建立数据库连接

执行基本查询

参数化查询 (防止 SQL 注入)

处理查询结果

处理大数据集 (使用游标)

执行 DML 语句 (增删改)

使用连接池 (提升性能)

完整代码示例

常见问题与解决方案

99ANYc3cd6

python sqlite3 多线程

python win32api 文档

Project教程2007怎么学？

Python开发技术详解PDF哪里能下载？

indesign教程cs5

Java Socket编程实例具体怎么实现？

Java中notify和wait的正确使用姿势是什么？

collections java

计算机二级Access教程怎么学？

CorelDraw自学教程从哪开始学？

Oracle存储过程如何与Java交互？

Python编程从入门到实践mobi适合零基础学吗？

iphone6 换屏教程

Thread join Java如何实现线程同步？

Java中如何准确判断Object的实际类型？

set to list java

Python如何高效查询Oracle数据库？

目录

环境准备

a. 安装 cx_Oracle 库

b. 安装 Oracle 客户端

建立数据库连接

执行基本查询

参数化查询 (防止 SQL 注入)

处理查询结果

处理大数据集 (使用游标)

执行 DML 语句 (增删改)

使用连接池 (提升性能)

完整代码示例

常见问题与解决方案

相关推荐

Java Socket编程实例具体怎么实现？

a. 安装 `cx_Oracle` 库