#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Oracle数据库连接测试脚本 用于测试Oracle数据库连接是否正常 """ import requests import json import oracledb from typing import Dict, Any import logging # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Oracle连接配置 ORACLE_CONFIG = { "db_type": "oracle", "host": "192.168.13.200", "port": 1521, "username": "bizuser", "password": "MySecurePass123", "database": "ORCLPDB1", # 服务名称 "mode": None, # 可以设置为 "SYSDBA", "SYSOPER" 等 "threaded": True, "extra_params": { # 可以添加其他Oracle特定参数 } } # API基础URL BASE_URL = "http://localhost:8000" def test_oracle_connection(config: Dict[str, Any]) -> bool: """ 测试Oracle数据库连接 Args: config: Oracle连接配置 Returns: bool: 连接是否成功 """ try: # 创建连接 print("正在测试Oracle连接...") print(f"连接信息: {config['host']}:{config['port']}/{config['database']}") print(f"用户名: {config['username']}") response = requests.post( f"{BASE_URL}/connections", json=config, headers={"Content-Type": "application/json"} ) if response.status_code == 200: result = response.json() if result.get("success"): connection_id = result["data"]["connection_id"] print(f"✅ Oracle连接成功! 连接ID: {connection_id}") # 测试查询 test_query(connection_id) # 关闭连接 close_connection(connection_id) return True else: print(f"❌ 连接失败: {result.get('message', '未知错误')}") if result.get('error'): print(f"错误详情: {result['error']}") return False else: print(f"❌ HTTP请求失败: {response.status_code}") print(f"响应内容: {response.text}") return False except Exception as e: print(f"❌ 连接测试异常: {str(e)}") return False def test_query(connection_id: str): """ 测试查询操作 Args: connection_id: 连接ID """ try: print("\n正在测试查询操作...") query_request = { "connection_id": connection_id, "sql": "SELECT 1 FROM DUAL" } response = requests.post( f"{BASE_URL}/query", json=query_request, headers={"Content-Type": "application/json"} ) if response.status_code == 200: result = response.json() if result.get("success"): print("✅ 查询测试成功!") print(f"查询结果: {result['data']}") else: print(f"❌ 查询失败: {result.get('message', '未知错误')}") else: print(f"❌ 查询请求失败: {response.status_code}") except Exception as e: print(f"❌ 查询测试异常: {str(e)}") def close_connection(connection_id: str): """ 关闭数据库连接 Args: connection_id: 连接ID """ try: print("\n正在关闭连接...") response = requests.delete( f"{BASE_URL}/connections/{connection_id}" ) if response.status_code == 200: result = response.json() if result.get("success"): print("✅ 连接已关闭") else: print(f"❌ 关闭连接失败: {result.get('message', '未知错误')}") else: print(f"❌ 关闭连接请求失败: {response.status_code}") except Exception as e: print(f"❌ 关闭连接异常: {str(e)}") def test_direct_oracle_connection(config: Dict[str, Any]) -> bool: """ 直接测试Oracle连接（不通过API） Args: config: Oracle连接配置 Returns: bool: 连接是否成功 """ try: print("\n🔍 直接测试Oracle连接...") # 方式1: 使用Easy Connect字符串 dsn1 = f"{config['host']}:{config['port']}/{config['database']}" print(f"尝试连接方式1 - Easy Connect: {dsn1}") connection = oracledb.connect( user=config['username'], password=config['password'], dsn=dsn1 ) # 测试查询 cursor = connection.cursor() cursor.execute("SELECT 1 FROM DUAL") result = cursor.fetchone() print(f"✅ 直接连接成功! 查询结果: {result}") # 获取数据库版本信息 cursor.execute("SELECT BANNER FROM V$VERSION WHERE ROWNUM = 1") version = cursor.fetchone() print(f"📊 数据库版本: {version[0] if version else 'Unknown'}") cursor.close() connection.close() return True except Exception as e: print(f"❌ 直接连接失败: {str(e)}") # 尝试其他连接方式 try: print("\n🔄 尝试其他连接方式...") # 方式2: 使用分离的参数 print(f"尝试连接方式2 - 分离参数: host={config['host']}, port={config['port']}, service_name={config['database']}") connection = oracledb.connect( user=config['username'], password=config['password'], host=config['host'], port=config['port'], service_name=config['database'] ) cursor = connection.cursor() cursor.execute("SELECT 1 FROM DUAL") result = cursor.fetchone() print(f"✅ 方式2连接成功! 查询结果: {result}") cursor.close() connection.close() return True except Exception as e2: print(f"❌ 方式2也失败: {str(e2)}") # 尝试使用SID而不是服务名 try: print("\n🔄 尝试使用SID连接...") # 方式3: 使用SID dsn3 = oracledb.makedsn(config['host'], config['port'], sid=config['database']) print(f"尝试连接方式3 - SID: {dsn3}") connection = oracledb.connect( user=config['username'], password=config['password'], dsn=dsn3 ) cursor = connection.cursor() cursor.execute("SELECT 1 FROM DUAL") result = cursor.fetchone() print(f"✅ SID连接成功! 查询结果: {result}") cursor.close() connection.close() return True except Exception as e3: print(f"❌ SID连接也失败: {str(e3)}") return False def main(): """ 主函数 """ print("=" * 60) print("Oracle数据库连接测试 - 增强版") print("=" * 60) # 首先进行直接连接测试 direct_success = test_direct_oracle_connection(ORACLE_CONFIG) if not direct_success: print("\n💥 直接Oracle连接失败!") print("\n🔧 可能的解决方案:") print("1. 检查Oracle服务是否正在运行") print("2. 检查网络连接和防火墙设置") print("3. 验证用户名、密码和服务名称") print("4. 确认Oracle客户端库已正确安装: pip install oracledb") print("5. 检查Oracle监听器配置: lsnrctl status") print("6. 尝试使用SID而不是服务名") print("7. 检查服务名是否正确注册到监听器") print("=" * 60) return # 检查API服务是否运行 print("\n🌐 检查API服务状态...") try: response = requests.get(f"{BASE_URL}/docs") if response.status_code != 200: print("❌ API服务未运行，请先启动服务: python main.py") return except requests.exceptions.ConnectionError: print("❌ 无法连接到API服务，请先启动服务: python main.py") return # 测试通过API的Oracle连接 print("\n🔗 测试通过API的Oracle连接...") api_success = test_oracle_connection(ORACLE_CONFIG) print("\n" + "=" * 60) if direct_success and api_success: print("🎉 所有Oracle连接测试通过!") elif direct_success: print("⚠️ 直接连接成功，但API连接失败") print("请检查API服务中的Oracle连接配置") else: print("💥 Oracle连接测试失败!") print("=" * 60) if __name__ == "__main__": main()