Files

47 lines
1.5 KiB
Python
Raw Permalink Normal View History

2026-09-13 16:19:24 +08:00
"""只读检查可用于 NL2SQL 真实验收的员工权限配置。"""
from __future__ import annotations
import asyncio
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from sqlalchemy import text
from config import database
from config.database.mysql import get_session_factory
async def main() -> None:
queries = {
"employees": text(
"SELECT id, employee_role, status FROM sys_user "
"WHERE user_type = 'EMPLOYEE' ORDER BY id LIMIT 50"
),
"roles": text(
"SELECT id, employee_role, can_query, max_rows, daily_quota, status "
"FROM nl2sql_query_role ORDER BY id"
),
"table_permissions": text(
"SELECT role_id, table_name, permission, status "
"FROM nl2sql_role_table_permission ORDER BY role_id, table_name LIMIT 200"
),
"column_permission_counts": text(
"SELECT role_id, table_name, COUNT(*) AS column_count "
"FROM nl2sql_role_column_permission WHERE status = 'active' "
"GROUP BY role_id, table_name ORDER BY role_id, table_name"
),
}
try:
async with get_session_factory()() as session:
for name, query in queries.items():
rows = [dict(row) for row in (await session.execute(query)).mappings()]
print(f"{name}: {rows}")
finally:
await database.mysql.dispose()
if __name__ == "__main__":
asyncio.run(main())