forked from Lesin/reposync
47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
import mysql.connector
|
||
import os
|
||
|
||
def test_mysql_connection():
|
||
try:
|
||
# 从环境变量获取配置
|
||
host = os.getenv('CEROBOT_MYSQL_HOST', '127.0.0.1')
|
||
port = int(os.getenv('CEROBOT_MYSQL_PORT', '3307'))
|
||
user = os.getenv('CEROBOT_MYSQL_USER', 'root')
|
||
password = os.getenv('CEROBOT_MYSQL_PWD', 'anye97!.!')
|
||
database = os.getenv('CEROBOT_MYSQL_DB', 'ob_sync')
|
||
|
||
print(f"尝试连接到MySQL服务器: {host}:{port}，数据库: {database}")
|
||
|
||
# 创建连接
|
||
connection = mysql.connector.connect(
|
||
host=host,
|
||
port=port,
|
||
user=user,
|
||
password=password,
|
||
database=database
|
||
)
|
||
|
||
if connection.is_connected():
|
||
db_info = connection.get_server_info()
|
||
print(f"成功连接到MySQL服务器，版本: {db_info}")
|
||
|
||
cursor = connection.cursor()
|
||
cursor.execute("SELECT DATABASE();")
|
||
database_name = cursor.fetchone()
|
||
print(f"当前数据库: {database_name[0]}")
|
||
|
||
cursor.execute("SHOW TABLES;")
|
||
tables = cursor.fetchall()
|
||
print("所有表:")
|
||
for table in tables:
|
||
print(table[0])
|
||
|
||
cursor.close()
|
||
connection.close()
|
||
print("MySQL连接已关闭")
|
||
|
||
except Exception as e:
|
||
print(f"连接失败: {str(e)}")
|
||
|
||
if __name__ == "__main__":
|
||
test_mysql_connection() |