292 lines
9.0 KiB
Python
292 lines
9.0 KiB
Python
#!/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() |