为了测试连接多种数据库，你需要使用对应的数据库驱动和API。以下是连接到各种数据库的基础代码示例。
MySQL:
import mysql.connector
config = {
'user': 'username',
'password': 'password',
'host': '127.0.0.1',
'database': 'database_name',
'raise_on_warnings': True
}
try:
connection = mysql.connector.connect(**config)
if connection.is_connected():
print("连接成功: MySQL")
# 这里可以执行其他数据库操作
except mysql.connector.Error as error:
print("连接失败: {}".format(error))
Doris:
import pymysql
config = {
'host': '127.0.0.1',
'user': 'username',
'password': 'password',
'database': 'database_name',
'port': 9030
}
try:
connection = pymysql.connect(**config)
if connection.open:
print("连接成功: Doris")
# 这里可以执行其他数据库操作
except pymysql.Error as error:
print("连接失败: {}".format(error))
Oracle:
import cx_Oracle
dsn = cx_Oracle.makedsn('host', 1521, 'service_name')
connection = cx_Oracle.connect(user='username', password='password', dsn=dsn)
if connection.session:
print("连接成功: Oracle")
# 这里可以执行其他数据库操作
else:
print("连接失败: {}".format(connection.exception()))
SQL Server:
import pymssql
connection = pymssql.connect(server='127.0.0.1', user='username', password='password', database='database_name')
if connection:
print("连接成功: SQL Server")
# 这里可以执行其他数据库操作
connection.close()
PostgreSQL:
import psycopg2
connection = psycopg2.connect(
dbname="database_name",
user="username",
password="password",
host="127.0.0.1",
port="5432"
)
if connection.is_closed:
print("连接成功: PostgreSQL")
# 这里可以执行其他数据库操作
connection.close()
Hive:
from pyhive import hive
conn = hive.Connection(host='127.0.0.1', port=10000, username='username')
if conn.open:
print("连接成功: Hive")
# 这里可以执行其他数据库操作
conn.close()
Elasticsearch:
from elasticsearch import Elasticsearch
es = Elasticsearch(hosts=['127.0.0.1:9200'])
if es.ping():
print("连接成功: Elasticsearch")
# 这里可以执行其他数据库操作
m 数据库连接示例未给出，因为需要具体到数据库类型和对应的Python库。通常，你需要安装对应的库（如pymongo用于MongoDB），然后使用类似下面的代码进行连接：
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
if client.server_info():
print("连接成功: MongoDB")
# 这里可以执行其