Python cx_Oracle连接数据库步骤有哪些？-杰瑞科技汇

cx_Oracle 是一个功能强大且广泛使用的 Python 扩展模块，它允许 Python 程序访问 Oracle 数据库。

（图片来源网络，侵删）

环境准备
- 安装 Oracle 客户端
- 安装 cx_Oracle Python 包
基本连接
- 简单连接
- 使用 DSN (Data Source Name)
- 处理连接池 (高级)
执行 SQL 语句
- 执行非查询语句 (INSERT, UPDATE, DELETE)
- 执行查询语句 (SELECT)
- 使用 with 语句自动管理游标
处理数据类型
- 基本数据类型
- BLOB 和 CLOB (重要)
最佳实践
- 使用连接池
- 使用 try...except...finally 进行错误处理
- 参数化查询 (防止 SQL 注入)
完整示例代码
常见问题与解决方案

环境准备

这是最关键的一步,很多人在这里遇到问题。

a. 安装 Oracle 客户端

cx_Oracle 本身不包含 Oracle 的网络库和客户端代码，它需要一个本地的 Oracle 客户端环境，你有两种主要选择：

选择 1: 安装完整的 Oracle 客户端 (传统方式)

从 Oracle 官网下载并安装 Instant Client。
Windows: 下载 ZIP 包，解压到一个固定目录（如 C:\oracle\instantclient_19_10），然后将该目录添加到系统的 PATH 环境变量中。
Linux: 下载 RPM (RedHat/CentOS) 或 DEB (Ubuntu/Debian) 包进行安装，安装后，通常会在 /usr/lib/oracle/版本号/client64/lib 等路径下，确保该路径在 LD_LIBRARY_PATH 环境变量中。
macOS: 使用 Homebrew 安装：brew install instantclient-basic

选择 2: 使用 cx_Oracle 内置的轻量级客户端 (推荐，更简单) 从 cx_Oracle 8.x 版本开始，它内置了一个轻量级的客户端，你不需要再单独安装 Oracle Instant Client，它会自动从 Oracle 的 Maven 仓库下载所需的库文件，这大大简化了部署过程。

（图片来源网络，侵删）

如何选择？

对于开发和测试,强烈推荐使用内置客户端。
对于生产环境,如果你的应用服务器已经安装了 Oracle 客户端，或者你有特定的版本控制要求，可以选择使用外部客户端。

b. 安装 `cx_Oracle` Python 包

使用 pip 进行安装，这是最简单的方式。

# 基础安装
pip install cx_Oracle
# 如果需要特定版本，可以指定
pip install cx_Oracle==8.3.0

基本连接

连接数据库需要用户名、密码和数据库连接信息。

a. 简单连接 (推荐)

这是最直接的方式,直接提供用户名、密码和 TNS (Transparent Network Substrate) 名称或 Easy Connect 字符串。

import cx_Oracle
# 连接信息
username = "your_username"
password = "your_password"
# TNS 名称 (在 tnsnames.ora 文件中定义) 或 Easy Connect 字符串
# Easy Connect 格式: hostname:port/service_name
dsn = "localhost:1521/ORCLPDB1" 
try:
    # 建立 connection 对象
    connection = cx_Oracle.connect(user=username, password=password, dsn=dsn)
    print("数据库连接成功!")
    print(f"数据库版本: {connection.version}")
except cx_Oracle.DatabaseError as e:
    error, = e.args
    print(f"数据库错误: {error.code} - {error.message}")
finally:
    # 确保连接被关闭
    if 'connection' in locals() and connection:
        connection.close()
        print("数据库连接已关闭。")

b. 使用 DSN (Data Source Name)

如果你有一个 tnsnames.ora 文件（通常在 Oracle 客户端安装目录的 network/admin 下），你可以使用其中定义的别名。

假设 tnsnames.ora 文件内容如下：

ORCL =
  (DESCRIPTION =
    (ADDRESS = (PROTOCOL = TCP)(HOST = localhost)(PORT = 1521))
    (CONNECT_DATA =
      (SERVER = DEDICATED)
      (SERVICE_NAME = ORCLPDB1)
    )
  )

Python 代码可以这样写：

import cx_Oracle
username = "your_username"
password = "your_password"
# 使用 tnsnames.ora 中定义的别名
dsn = "ORCL" 
connection = cx_Oracle.connect(user=username, password=password, dsn=dsn)
# ... 后续操作

执行 SQL 语句

连接成功后,你需要创建一个 cursor 对象来执行 SQL 语句。

a. 执行非查询语句 (DML: INSERT, UPDATE, DELETE)

这类语句会改变数据库中的数据,通常需要执行 connection.commit() 来提交事务。

import cx_Oracle
# ... (连接代码同上) ...
try:
    connection = cx_Oracle.connect(user="hr", password="your_password", dsn="localhost:1521/ORCLPDB1")
    cursor = connection.cursor()
    # 准备 SQL 语句 (使用参数化查询，见最佳实践部分)
    sql = "INSERT INTO employees (employee_id, first_name, last_name, email) VALUES (:1, :2, :3, :4)"
    employee_data = (101, 'John', 'Doe', 'john.doe@example.com')
    # 执行 SQL
    cursor.execute(sql, employee_data)
    # 提交事务
    connection.commit()
    print(f"成功插入 {cursor.rowcount} 行数据。")
except cx_Oracle.DatabaseError as e:
    # 发生错误时回滚
    if 'connection' in locals() and connection:
        connection.rollback()
    error, = e.args
    print(f"数据库错误: {error.code} - {error.message}")
finally:
    if 'cursor' in locals() and cursor:
        cursor.close()
    if 'connection' in locals() and connection:
        connection.close()

b. 执行查询语句 (SELECT)

查询语句会返回结果集,你需要遍历游标来获取数据。

import cx_Oracle
# ... (连接代码同上) ...
try:
    connection = cx_Oracle.connect(user="hr", password="your_password", dsn="localhost:1521/ORCLPDB1")
    cursor = connection.cursor()
    sql = "SELECT employee_id, first_name, last_name FROM employees WHERE department_id = :1"
    department_id = 50
    # 执行 SQL
    cursor.execute(sql, (department_id,)) # 注意参数是元组
    # 遍历结果
    print(f"部门 {department_id} 的员工列表:")
    for row in cursor:
        # row 是一个元组，如 (100, 'Steven', 'King')
        print(f"ID: {row[0]}, 姓名: {row[1]} {row[2]}")
    # 或者一次性获取所有结果到列表中 (对于大数据集要小心)
    # all_rows = cursor.fetchall()
    # for row in all_rows:
    #     print(row)
except cx_Oracle.DatabaseError as e:
    error, = e.args
    print(f"数据库错误: {error.code} - {error.message}")
finally:
    if 'cursor' in locals() and cursor:
        cursor.close()
    if 'connection' in locals() and connection:
        connection.close()

c. 使用 `with` 语句自动管理游标

cursor 对象支持上下文管理器协议，可以使用 with 语句，这样在 with 块结束时，游标会自动关闭，非常方便。

import cx_Oracle
# ... (连接代码同上) ...
try:
    connection = cx_Oracle.connect(user="hr", password="your_password", dsn="localhost:1521/ORCLPDB1")
    with connection.cursor() as cursor:
        sql = "SELECT first_name, last_name FROM employees WHERE rownum < 5"
        cursor.execute(sql)
        print("前4名员工:")
        for first_name, last_name in cursor: # 可以直接解包
            print(f"- {first_name} {last_name}")
    # 退出 with 块后，cursor 会自动关闭
except cx_Oracle.DatabaseError as e:
    error, = e.args
    print(f"数据库错误: {error.code} - {error.message}")
finally:
    if 'connection' in locals() and connection:
        connection.close()

处理数据类型

cx_Oracle 会自动处理大部分基本数据类型的转换，但 BLOB 和 CLOB 需要特别处理。

BLOB (Binary Large Object) 和 CLOB (Character Large Object)

读取 BLOB/CLOB:

# ... (连接和游标代码) ...
sql = "SELECT image_blob, resume_clob FROM documents WHERE doc_id = 1"
cursor.execute(sql)
# 使用 fetchone() 获取单行
row = cursor.fetchone()
if row:
    blob_data = row[0] # BLOB 数据以 bytes 类型返回
    clob_data = row[1] # CLOB 数据以 str 类型返回
    # 将 BLOB 数据写入文件
    with open("my_image.png", "wb") as f:
        f.write(blob_data)
    # 直接打印 CLOB 内容 (注意大小)
    print(clob_data[:200] + "...") # 只打印前200个字符

写入 BLOB/CLOB: 写入 LOB 数据不能直接通过 cursor.execute() 完成，需要使用 LOB 对象。

import cx_Oracle
import io
# ... (连接代码) ...
try:
    connection = cx_Oracle.connect(user="hr", password="your_password", dsn="localhost:1521/ORCLPDB1")
    cursor = connection.cursor()
    # 1. 插入一个空 LOB，并获取其 LOB 对象
    sql = "INSERT INTO documents (doc_id, doc_name, image_blob) VALUES (seq_docs.NEXTVAL, :1, EMPTY_BLOB()) RETURNING image_blob INTO :2"
    doc_name = "Test Image"
    # 创建一个变量来接收返回的 LOB 对象
    lob_var = cursor.var(cx_Oracle.LOB)
    cursor.execute(sql, (doc_name, lob_var))
    # 2. 获取 LOB 对象并写入数据
    lob = lob_var.getvalue()
    # 假设你有一些二进制数据
    image_data = b"这是一些模拟的图片二进制数据"
    # 打开 LOB 进行写入
    lob.write(image_data)
    connection.commit()
    print("BLOB 数据写入成功。")
except cx_Oracle.DatabaseError as e:
    if 'connection' in locals() and connection:
        connection.rollback()
    error, = e.args
    print(f"数据库错误: {error.code} - {error.message}")
finally:
    if 'cursor' in locals() and cursor:
        cursor.close()
    if 'connection' in locals() and connection:
        connection.close()

最佳实践

a. 使用连接池

在高并发应用中,为每个请求创建和销毁连接是非常消耗资源的，连接池可以复用已建立的数据库连接，显著提高性能。

import cx_Oracle
# 连接池配置
pool = cx_Oracle.SessionPool(
    user="hr",
    password="your_password",
    dsn="localhost:1521/ORCLPDB1",
    min=1,      # 连接池中最少连接数
    max=2,      # 连接池中最大连接数
    increment=1, # 当连接不够时，一次增加的连接数
    threaded=True
)
try:
    # 从连接池中获取一个连接
    with pool.acquire() as connection:
        with connection.cursor() as cursor:
            cursor.execute("SELECT * FROM my_table")
            for row in cursor:
                print(row)
    # 连接会在 with 块结束时自动返回到连接池
except cx_Oracle.DatabaseError as e:
    error, = e.args
    print(f"数据库错误: {error.code} - {error.message}")
finally:
    # 应用关闭时，关闭连接池
    pool.close()

b. 使用 `try...except...finally` 进行错误处理

如上所有示例所示,必须妥善处理数据库操作中可能发生的异常，并确保资源（游标、连接）被正确释放。

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

永远不要使用 Python 的字符串拼接来构建 SQL 查询，这会导致严重的 SQL 注入漏洞。

错误示范 (危险!):

user_id = input("请输入用户ID: ")
sql = f"SELECT * FROM users WHERE id = {user_id}"
cursor.execute(sql) # 恶意用户输入 "1; DROP TABLE users; --" 将导致灾难

正确示范 (安全): cx_Oracle 使用命名占位符 name 或位置占位符 1, 2。

user_id = 101
sql = "SELECT * FROM users WHERE id = :1" # 使用 :1, :2 ... 是位置占位符
# 或者
# sql = "SELECT * FROM users WHERE id = :user_id" # 使用 :name 是命名占位符
cursor.execute(sql, (user_id,)) # 参数必须是元组或字典
# 如果使用命名占位符，也可以这样传参:
# cursor.execute(sql, {'user_id': user_id})

完整示例代码

这个例子综合了连接、查询、参数化查询和错误处理。

import cx_Oracle
import sys
def get_employees_by_department(connection, department_id):
    """查询指定部门的员工信息"""
    try:
        with connection.cursor() as cursor:
            # 使用命名参数
            sql = """
                SELECT employee_id, first_name, last_name, job_id
                FROM employees
                WHERE department_id = :dept_id
                ORDER BY last_name
            """
            cursor.execute(sql, dept_id=department_id)
            print(f"\n--- 部门 {department_id} 的员工 ---")
            if not cursor:
                print("未找到任何员工。")
                return
            for emp_id, first, last, job in cursor:
                print(f"ID: {emp_id:<5} | 姓名: {last}, {first:<10} | 职位: {job}")
    except cx_Oracle.DatabaseError as e:
        error, = e.args
        print(f"查询员工时出错: {error.code} - {error.message}")
        raise # 重新抛出异常，由上层处理
def main():
    # 假设的数据库连接信息
    db_user = "hr"
    db_password = "your_password" # 请替换为你的密码
    db_dsn = "localhost:1521/ORCLPDB1" # 请替换为你的DSN
    connection = None
    try:
        # 1. 连接数据库
        print("正在尝试连接数据库...")
        connection = cx_Oracle.connect(user=db_user, password=db_password, dsn=db_dsn)
        print("连接成功!")
        print(f"数据库版本: {connection.version}")
        # 2. 调用函数查询数据
        get_employees_by_department(connection, department_id=50)
        get_employees_by_department(connection, department_id=80)
    except cx_Oracle.DatabaseError as e:
        error, = e.args
        print(f"\n发生严重的数据库错误: {error.code} - {error.message}", file=sys.stderr)
    except Exception as e:
        print(f"\n发生未知错误: {e}", file=sys.stderr)
    finally:
        # 3. 确保连接被关闭
        if connection:
            connection.close()
            print("\n数据库连接已关闭。")
if __name__ == "__main__":
    main()

常见问题与解决方案

Error: DPI-1047: Cannot locate a 64-bit Oracle Client library
- 原因: 你在 64 位 Python 环境中运行，但系统找不到 64 位的 Oracle 客户端库。
- 解决方案:
  - (推荐) 确保你安装了 64 位的 Instant Client。
  - 如果使用 cx_Oracle 内置客户端，确保你的网络可以访问 Maven 仓库。
  - 检查 PATH (Windows) 或 LD_LIBRARY_PATH (Linux) 环境变量是否正确指向了 Instant Client 的 lib 目录。
ORA-12154: TNS:could not resolve the connect identifier specified
- 原因: dsn 参数中的服务名或 TNS 别名无法解析。
- 解决方案:
  - 检查 dsn 字符串是否正确，确认 Oracle 服务的 hostname, port, 和 service_name 或 SID。
  - 如果使用 TNS 别名，确认 tnsnames.ora 文件存在，并且别名拼写正确，且文件路径在 TNS_ADMIN 环境变量中或客户端的默认位置。
ORA-01017: invalid username/password; logon denied
- 原因: 用户名或密码错误。
- 解决方案: 仔细检查用户名和密码是否正确，注意大小写。
Unicode 编码问题 (Python 2 或旧版客户端)
- 原因: 在处理非英文字符时出现乱码。
- 解决方案 (Python 3): Python 3 的字符串默认是 Unicode，cx_Oracle 也默认使用 Unicode，所以通常不会有问题，确保你的数据库字符集（如 AL32UTF8）与你的应用环境兼容即可。

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

Python cx_Oracle连接数据库步骤有哪些？

目录

环境准备

a. 安装 Oracle 客户端

b. 安装 `cx_Oracle` Python 包

基本连接

a. 简单连接 (推荐)

b. 使用 DSN (Data Source Name)

执行 SQL 语句

a. 执行非查询语句 (DML: INSERT, UPDATE, DELETE)

b. 执行查询语句 (SELECT)

c. 使用 `with` 语句自动管理游标

处理数据类型

BLOB (Binary Large Object) 和 CLOB (Character Large Object)

最佳实践

a. 使用连接池

b. 使用 `try...except...finally` 进行错误处理

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

完整示例代码

常见问题与解决方案

99ANYc3cd6

Python中如何判断NoneType对象？

新概念英语视频教程哪里能免费下载？

Java class如何正确调用方法？

微信小程序直播开发视频教程

note2自换外屏玻璃教程

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

Java socket阻塞模式如何高效使用？

Java内部类如何调用外部类变量？

collections java

Photoshop 5教程从哪开始学？

Python re match 返回什么类型？

wap无限流量卡如何走4G？

Think in Python中文版适合谁学？

如何将单个Python文件转为可安装的Python包？

Java Map如何高效转为Bean对象？

哪里能找到byte of pythonPDF版？

Python cx_Oracle连接数据库步骤有哪些？

目录

环境准备

a. 安装 Oracle 客户端

b. 安装 cx_Oracle Python 包

基本连接

a. 简单连接 (推荐)

b. 使用 DSN (Data Source Name)

执行 SQL 语句

a. 执行非查询语句 (DML: INSERT, UPDATE, DELETE)

b. 执行查询语句 (SELECT)

c. 使用 with 语句自动管理游标

处理数据类型

BLOB (Binary Large Object) 和 CLOB (Character Large Object)

最佳实践

a. 使用连接池

b. 使用 try...except...finally 进行错误处理

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

完整示例代码

常见问题与解决方案

相关推荐

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

b. 安装 `cx_Oracle` Python 包

c. 使用 `with` 语句自动管理游标

b. 使用 `try...except...finally` 进行错误处理