67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
|
|
#!/usr/bin/env python3
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
测试数据库连接和表结构
|
||
|
|
"""
|
||
|
|
import sys
|
||
|
|
import os
|
||
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
|
||
|
|
from src.flask_prompt_master import create_app, db
|
||
|
|
from sqlalchemy import text
|
||
|
|
|
||
|
|
def test_database():
|
||
|
|
"""测试数据库连接"""
|
||
|
|
app = create_app()
|
||
|
|
|
||
|
|
with app.app_context():
|
||
|
|
print("=" * 60)
|
||
|
|
print("测试数据库连接")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 测试数据库连接
|
||
|
|
result = db.session.execute(text("SELECT 1")).fetchone()
|
||
|
|
print("✅ 数据库连接成功")
|
||
|
|
|
||
|
|
# 检查数据库名称
|
||
|
|
db_name = db.session.execute(text("SELECT DATABASE()")).fetchone()[0]
|
||
|
|
print("📊 当前数据库: {}".format(db_name))
|
||
|
|
|
||
|
|
# 检查表列表
|
||
|
|
tables = db.session.execute(text("SHOW TABLES")).fetchall()
|
||
|
|
print("\n📋 数据库表列表:")
|
||
|
|
for table in tables:
|
||
|
|
print(" - {}".format(table[0]))
|
||
|
|
|
||
|
|
# 检查user表是否存在
|
||
|
|
if any('user' in str(table) for table in tables):
|
||
|
|
print("\n✅ user表存在")
|
||
|
|
|
||
|
|
# 检查user表结构
|
||
|
|
columns = db.session.execute(text("DESCRIBE user")).fetchall()
|
||
|
|
print("\n📋 user表结构:")
|
||
|
|
for col in columns:
|
||
|
|
print(" {}: {}".format(col[0], col[1]))
|
||
|
|
|
||
|
|
# 检查user表数据
|
||
|
|
count = db.session.execute(text("SELECT COUNT(*) FROM user")).fetchone()[0]
|
||
|
|
print("\n📊 user表记录数: {}".format(count))
|
||
|
|
|
||
|
|
if count > 0:
|
||
|
|
# 显示前几条记录
|
||
|
|
records = db.session.execute(text("SELECT * FROM user LIMIT 3")).fetchall()
|
||
|
|
print("\n📝 前3条记录:")
|
||
|
|
for i, record in enumerate(records, 1):
|
||
|
|
print(" 记录{}: {}".format(i, dict(record._mapping)))
|
||
|
|
else:
|
||
|
|
print("\n❌ user表不存在")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print("❌ 数据库连接失败: {}".format(str(e)))
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
test_database()
|