47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
"""只读检查可用于 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())
|