40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
|
"""Read-only check for投顾 Agent/工作台 MySQL tables and Redis connectivity."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from sqlalchemy import text
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
|
|
||
|
|
from config.database import mysql, redis
|
||
|
|
|
||
|
|
|
||
|
|
async def check() -> None:
|
||
|
|
try:
|
||
|
|
async with mysql.get_engine().connect() as connection:
|
||
|
|
for table in (
|
||
|
|
"advisor_draft",
|
||
|
|
"event_log",
|
||
|
|
"sensitive_word",
|
||
|
|
"advisor_report",
|
||
|
|
"advisor_todo",
|
||
|
|
"advisor_visit_record",
|
||
|
|
):
|
||
|
|
result = await connection.execute(
|
||
|
|
text("SHOW TABLES LIKE :table"), {"table": table}
|
||
|
|
)
|
||
|
|
print(f"{table}: {'present' if result.first() else 'missing'}")
|
||
|
|
print("mysql: ok")
|
||
|
|
await redis.client().ping()
|
||
|
|
print("redis: ok")
|
||
|
|
finally:
|
||
|
|
await mysql.dispose()
|
||
|
|
await redis.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(check())
|