### 使用Python连接Oracle数据库的方法及示例代码 为了实现通过Python连接到Oracle数据库的功能，通常会依赖于`cx_Oracle`库。这是一个专门设计用于与Oracle数据库交互的扩展模块。以下是详细的说明以及完整的示例代码。 #### 1. 安装 `cx_Oracle` 库 在使用之前，需要安装`cx_Oracle`库。可以通过以下命令完成安装： ```bash pip install cx_Oracle ``` #### 2. 示例代码：基本连接与查询 下面展示了一个简单的例子，演示如何连接到Oracle数据库并执行基础查询操作。 ```python import cx_Oracle # 导入cx_Oracle库以支持Oracle数据库连接[^1] # 定义连接参数 username = "your_username" password = "your_password" dsn = "hostname:port/service_name" try: # 建立数据库连接 connection = cx_Oracle.connect(username, password, dsn) # 创建连接对象[^1] print("成功连接至数据库:", connection.version) # 创建游标对象以便执行SQL语句 cursor = connection.cursor() # 编写SQL查询语句 sql_query = "SELECT * FROM your_table_name" # 替换为实际表名[^1] # 执行查询 cursor.execute(sql_query) # 执行SQL查询 # 获取所有结果 rows = cursor.fetchall() # 取回全部查询结果 for row in rows: print(row) # 输出每一行的数据 except cx_Oracle.Error as error: print("发生错误:", error) # 处理可能出现的异常情况 finally: # 确保资源被释放 if cursor: cursor.close() # 关闭游标 if connection: connection.close() # 断开数据库连接[^1] ``` --- #### 3. 添加超时处理机制 (高级用法) 如果希望进一步提升程序健壮性，可以在查询过程中引入超时控制逻辑。这有助于防止长时间未响应的操作阻塞整个应用程序流程。 ```python import cx_Oracle from threading import Thread import queue import pandas as pd def run_sql_oracle(sql: str) -> pd.DataFrame or str or None: global conn # 假设已存在全局变量conn表示预定义好的数据库连接[^3] if not conn: return "无有效数据库连接." result_queue = queue.Queue() # 初始化队列存储查询结果 def run_query(q): try: cur = conn.cursor() cur.execute(sql) columns = [col[0] for col in cur.description] # 获取字段名称列表 data = cur.fetchall() df = pd.DataFrame(data, columns=columns) # 构建DataFrame结构化数据 q.put(df) # 将结果放入队列中 except Exception as e: q.put(e) # 如果有异常，则将异常信息存入队列 finally: cur.close() thread = Thread(target=run_query, args=(result_queue,)) thread.start() timeout = 15 # 设置最大等待时间为15秒 try: results = result_queue.get(timeout=timeout) # 阻塞直到获得结果或者达到超时时间[^3] if isinstance(results, Exception): # 判断取出的内容是否属于异常类 print(f"查询失败: {results}") return f"{results}" else: return results except queue.Empty: # 若超过指定时限仍未收到任何消息 print("查询超时.") return "查询超时." finally: thread.join() # 主线程等待子线程结束后再继续往下走[^3] if __name__ == "__main__": sample_sql = "SELECT column1, column2 FROM some_table LIMIT 10;" outcome = run_sql_oracle(sample_sql) if isinstance(outcome, pd.DataFrame): print("查询结果:") print(outcome.head()) elif type(outcome) is str and ("失败" in outcome or "超时" in outcome): print("出现问题:", outcome) ``` ---