chenqi
4a46e7d2f5
- 将默认数据库用户名从 your_user 改为 postgresql，明确用户名字段说明 - 修改 .env.example 中 DEBUG 默认为 True，方便本地调试 - 更新 deploy-docker-compose.sh 脚本中 DEBUG 默认值为 True，保持一致性 - 保持数据库连接字符串格式正确，确保环境变量配置一致
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""
|
|
数据库连接测试脚本
|
|
"""
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 添加项目根目录到路径
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
|
|
async def test_connection():
|
|
"""测试数据库连接"""
|
|
print("=" * 50)
|
|
print("数据库连接诊断")
|
|
print("=" * 50)
|
|
|
|
# 1. 显示当前工作目录
|
|
print(f"\n1. 当前工作目录: {os.getcwd()}")
|
|
|
|
# 2. 检查 .env 文件
|
|
env_path = Path(__file__).parent / ".env"
|
|
print(f"2. .env 文件路径: {env_path}")
|
|
print(f" .env 文件存在: {env_path.exists()}")
|
|
|
|
if env_path.exists():
|
|
print(f" .env 文件内容:")
|
|
with open(env_path) as f:
|
|
for line in f:
|
|
if "PASSWORD" in line.upper() or "SECRET" in line.upper():
|
|
# 隐藏敏感信息
|
|
print(f" {line.strip()[:30]}***")
|
|
else:
|
|
print(f" {line.strip()}")
|
|
|
|
# 3. 加载配置
|
|
print("\n3. 加载配置:")
|
|
from app.core.config import settings
|
|
print(f" DATABASE_URL: {settings.DATABASE_URL[:50]}***")
|
|
print(f" DEBUG: {settings.DEBUG}")
|
|
|
|
# 4. 测试数据库连接
|
|
print("\n4. 测试数据库连接:")
|
|
try:
|
|
engine = create_async_engine(settings.DATABASE_URL, echo=True)
|
|
async with engine.connect() as conn:
|
|
result = await conn.execute(text("SELECT 1"))
|
|
print(" ✅ 数据库连接成功!")
|
|
await engine.dispose()
|
|
except Exception as e:
|
|
print(f" ❌ 数据库连接失败: {type(e).__name__}: {e}")
|
|
|
|
# 5. 测试其他可能的连接字符串
|
|
print("\n5. 测试备选连接字符串:")
|
|
test_urls = [
|
|
"postgresql+asyncpg://postgresql:Jchl1528@192.168.110.252:15432/hospital_performance",
|
|
"postgresql+asyncpg://postgres:Jchl1528@192.168.110.252:15432/hospital_performance",
|
|
]
|
|
|
|
for url in test_urls:
|
|
try:
|
|
engine = create_async_engine(url)
|
|
async with engine.connect() as conn:
|
|
result = await conn.execute(text("SELECT 1"))
|
|
print(f" ✅ 连接成功: {url[:40]}***")
|
|
await engine.dispose()
|
|
break
|
|
except Exception as e:
|
|
print(f" ❌ 失败: {str(e)[:50]}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_connection())
|