first commit
This commit is contained in:
292
test_oracle_connection.py
Normal file
292
test_oracle_connection.py
Normal file
@@ -0,0 +1,292 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user