36 lines
1.0 KiB
Python
36 lines
1.0 KiB
Python
#!/usr/bin/env python3
|
|
"""检查测试表单数据"""
|
|
|
|
import asyncio
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy import text
|
|
|
|
DB_CONFIG = {
|
|
'host': '123.57.81.127',
|
|
'port': 3306,
|
|
'user': 'cpy_admin',
|
|
'password': 'Tgzz2025+',
|
|
'database': 'ruoyi-fastapi'
|
|
}
|
|
|
|
DATABASE_URL = (
|
|
f"mysql+asyncmy://{DB_CONFIG['user']}:{DB_CONFIG['password']}@"
|
|
f"{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}"
|
|
)
|
|
|
|
async def check():
|
|
engine = create_async_engine(DATABASE_URL, echo=False)
|
|
async with engine.connect() as conn:
|
|
result = await conn.execute(text("""
|
|
SELECT id, name, form_type, is_active, del_flag, LENGTH(form_json) as json_len
|
|
FROM test_form ORDER BY id
|
|
"""))
|
|
rows = result.fetchall()
|
|
print(f"Total forms: {len(rows)}")
|
|
for row in rows:
|
|
print(f" ID:{row[0]} | {row[1]} | Type:{row[2]} | Active:{row[3]} | Del:{row[4]} | JsonLen:{row[5]}")
|
|
await engine.dispose()
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(check())
|