73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
测试管理员查询
|
|
运行: python test_admin_query.py
|
|
"""
|
|
|
|
from sqlalchemy import create_engine, text
|
|
|
|
# 数据库连接
|
|
engine = create_engine(
|
|
"mysql+pymysql://root:Zzl33818!@127.0.0.1:3306/stock?charset=utf8mb4"
|
|
)
|
|
|
|
def test_admin_query():
|
|
print("=" * 50)
|
|
print("测试管理员查询")
|
|
print("=" * 50)
|
|
|
|
with engine.connect() as conn:
|
|
# 1. 查看表是否存在
|
|
print("\n1. 检查 community_admins 表...")
|
|
try:
|
|
result = conn.execute(text("SHOW TABLES LIKE 'community_admins'")).fetchone()
|
|
if result:
|
|
print(" ✓ 表存在")
|
|
else:
|
|
print(" ✗ 表不存在！")
|
|
return
|
|
except Exception as e:
|
|
print(f" ✗ 错误: {e}")
|
|
return
|
|
|
|
# 2. 查看表中所有数据
|
|
print("\n2. 查看 community_admins 表中所有数据...")
|
|
try:
|
|
result = conn.execute(text("SELECT * FROM community_admins")).fetchall()
|
|
print(f" 共 {len(result)} 条记录:")
|
|
for row in result:
|
|
print(f" - id={row.id}, user_id={row.user_id}, role={row.role}")
|
|
except Exception as e:
|
|
print(f" ✗ 错误: {e}")
|
|
|
|
# 3. 查询 user_id=65
|
|
print("\n3. 查询 user_id=65...")
|
|
try:
|
|
sql = text("SELECT role, permissions FROM community_admins WHERE user_id = :user_id")
|
|
result = conn.execute(sql, {'user_id': 65}).fetchone()
|
|
if result:
|
|
print(f" ✓ 找到记录: role={result.role}, permissions={result.permissions}")
|
|
else:
|
|
print(" ✗ 未找到 user_id=65 的记录！")
|
|
except Exception as e:
|
|
print(f" ✗ 错误: {e}")
|
|
|
|
# 4. 查询 user_id='65' (字符串)
|
|
print("\n4. 查询 user_id='65' (字符串形式)...")
|
|
try:
|
|
sql = text("SELECT role, permissions FROM community_admins WHERE user_id = :user_id")
|
|
result = conn.execute(sql, {'user_id': '65'}).fetchone()
|
|
if result:
|
|
print(f" ✓ 找到记录: role={result.role}")
|
|
else:
|
|
print(" ✗ 未找到记录")
|
|
except Exception as e:
|
|
print(f" ✗ 错误: {e}")
|
|
|
|
print("\n" + "=" * 50)
|
|
print("测试完成")
|
|
print("=" * 50)
|
|
|
|
if __name__ == '__main__':
|
|
test_admin_query()
|