Files
aitsc/tests/test_db_connection.py
renjianbo 3b806e2bde refactor: project restructuring - migrate to modular directory layout
- Move root-level docs into docs/ directory
- Move config files into config/ directory
- Move docker files into docker/ directory
- Move test scripts into tests/ directory
- Remove .env from tracking (use .env.example as template)
- Remove .venv/ from tracking (use requirements.txt)
- Add Vue3 frontend app (vue-app/)
- Add new routes: upload, user_templates, meeting_minutes, etc.
- Add database migrations for prompt_template additions
- Fix load_dotenv() to use absolute path for Flask reloader compatibility

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-06-13 22:34:53 +08:00

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()