106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
|
"""NL2SQL 管理能力服务。"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from model.nl2sql_permission import (
|
||
|
|
Nl2SqlQueryRole,
|
||
|
|
Nl2SqlRoleColumnPermission,
|
||
|
|
Nl2SqlRoleTablePermission,
|
||
|
|
Nl2SqlSensitiveField,
|
||
|
|
)
|
||
|
|
from repositories.nl2sql_permission import Nl2SqlPermissionRepo
|
||
|
|
|
||
|
|
VALID_MASK_TYPES = {"partial", "hash"}
|
||
|
|
VALID_ROW_SCOPE_TYPES = {"none", "customer_ids", "product_ids"}
|
||
|
|
|
||
|
|
|
||
|
|
def _clean_text(value: str, name: str) -> str:
|
||
|
|
cleaned = value.strip()
|
||
|
|
if not cleaned:
|
||
|
|
raise ValueError(f"{name}不能为空")
|
||
|
|
return cleaned
|
||
|
|
|
||
|
|
|
||
|
|
def validate_mask_type(mask_type: str | None) -> str | None:
|
||
|
|
if mask_type is not None and mask_type not in VALID_MASK_TYPES:
|
||
|
|
raise ValueError("脱敏类型必须为 partial 或 hash")
|
||
|
|
return mask_type
|
||
|
|
|
||
|
|
|
||
|
|
def validate_table_permission(payload: dict) -> dict:
|
||
|
|
permission = payload.get("permission", "SELECT")
|
||
|
|
if permission != "SELECT":
|
||
|
|
raise ValueError("表权限只允许 SELECT")
|
||
|
|
row_scope_type = payload.get("row_scope_type", "none")
|
||
|
|
if row_scope_type not in VALID_ROW_SCOPE_TYPES:
|
||
|
|
raise ValueError("行级范围类型无效")
|
||
|
|
return {
|
||
|
|
"table_name": _clean_text(payload["table_name"], "表名"),
|
||
|
|
"permission": "SELECT",
|
||
|
|
"row_scope_type": row_scope_type,
|
||
|
|
"row_scope_column": payload.get("row_scope_column"),
|
||
|
|
"status": payload.get("status", "active"),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
async def create_role(db, payload: dict, *, repo_factory=Nl2SqlPermissionRepo):
|
||
|
|
values = {
|
||
|
|
"role_code": _clean_text(payload["role_code"], "角色编码"),
|
||
|
|
"role_name": _clean_text(payload["role_name"], "角色名称"),
|
||
|
|
"employee_role": _clean_text(payload["employee_role"], "员工角色"),
|
||
|
|
"can_query": bool(payload.get("can_query", False)),
|
||
|
|
"max_rows": int(payload.get("max_rows", 1000)),
|
||
|
|
"daily_quota": int(payload.get("daily_quota", 0)),
|
||
|
|
"status": "active",
|
||
|
|
}
|
||
|
|
if values["max_rows"] <= 0 or values["daily_quota"] < 0:
|
||
|
|
raise ValueError("配额参数无效")
|
||
|
|
return await repo_factory(db).add_role(**values)
|
||
|
|
|
||
|
|
|
||
|
|
def role_payload(role: Nl2SqlQueryRole) -> dict:
|
||
|
|
return {
|
||
|
|
"id": role.id,
|
||
|
|
"role_code": role.role_code,
|
||
|
|
"role_name": role.role_name,
|
||
|
|
"employee_role": role.employee_role,
|
||
|
|
"can_query": role.can_query,
|
||
|
|
"max_rows": role.max_rows,
|
||
|
|
"daily_quota": role.daily_quota,
|
||
|
|
"status": role.status,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def table_permission_payload(item: Nl2SqlRoleTablePermission) -> dict:
|
||
|
|
return {
|
||
|
|
"id": item.id,
|
||
|
|
"role_id": item.role_id,
|
||
|
|
"table_name": item.table_name,
|
||
|
|
"permission": item.permission,
|
||
|
|
"row_scope_type": item.row_scope_type,
|
||
|
|
"row_scope_column": item.row_scope_column,
|
||
|
|
"status": item.status,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def column_permission_payload(item: Nl2SqlRoleColumnPermission) -> dict:
|
||
|
|
return {
|
||
|
|
"id": item.id,
|
||
|
|
"role_id": item.role_id,
|
||
|
|
"table_name": item.table_name,
|
||
|
|
"column_name": item.column_name,
|
||
|
|
"access_mode": item.access_mode,
|
||
|
|
"mask_type": item.mask_type,
|
||
|
|
"status": item.status,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def sensitive_field_payload(item: Nl2SqlSensitiveField) -> dict:
|
||
|
|
return {
|
||
|
|
"id": item.id,
|
||
|
|
"table_name": item.table_name,
|
||
|
|
"column_name": item.column_name,
|
||
|
|
"mask_type": item.mask_type,
|
||
|
|
"description": item.description,
|
||
|
|
"status": item.status,
|
||
|
|
}
|