klinsmaya/SQLBot
1
0
Fork 0
mirror of https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git synced 2026-11-08 01:06:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: permission json to text

Browse Source
This commit is contained in:
junjun
2025-08-05 10:37:35 +08:00
parent 49a9f55306
commit 25132db697
4 changed files with 53 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/datasource/crud/datasource.py
+2 -2
View File
@@ -25,8 +25,8 @@ def get_datasource_list(session: SessionDep, user: CurrentUser, oid: Optional[in
current_oid = user.oid if user.oid is not None else 1
if user.isAdmin and oid:
current_oid = oid
return session.exec(select(CoreDatasource).where(CoreDatasource.oid == current_oid).order_by(
func.convert_to(CoreDatasource.name, 'gbk'))).all()
return session.exec(
select(CoreDatasource).where(CoreDatasource.oid == current_oid).order_by(CoreDatasource.name)).all()
def get_ds(session: SessionDep, id: int):
backend/apps/datasource/crud/permission.py
+50 -27
View File
@@ -1,4 +1,3 @@
import json
from typing import List, Optional
@@ -11,56 +10,80 @@ from sqlbot_xpack.permissions.models.ds_permission import DsPermission, Permissi
from sqlbot_xpack.permissions.models.ds_rules import DsRules
from sqlalchemy.dialects.postgresql import JSONB
def get_row_permission_filters(session: SessionDep, current_user: CurrentUser, ds: CoreDatasource, tables: Optional[list] = None, single_table: Optional[CoreTable] = None):
def get_row_permission_filters(session: SessionDep, current_user: CurrentUser, ds: CoreDatasource,
tables: Optional[list] = None, single_table: Optional[CoreTable] = None):
if single_table:
table_list = [session.get(CoreTable, single_table.id)]
else:
table_list =session.query(CoreTable).filter(
table_list = session.query(CoreTable).filter(
and_(CoreTable.ds_id == ds.id, CoreTable.table_name.in_(tables))
).all()
filters = []
for table in table_list:
row_permissions = session.query(DsPermission).filter(
and_(DsPermission.table_id == table.id, DsPermission.type == 'row')).all()
res: List[PermissionDTO] = []
if row_permissions is not None:
for permission in row_permissions:
# check permission and user in same rules
obj = session.query(DsRules).filter(
and_(DsRules.permission_list.op('@>')(cast([permission.id], JSONB)),
or_(DsRules.user_list.op('@>')(cast([f'{current_user.id}'], JSONB)),
DsRules.user_list.op('@>')(cast([current_user.id], JSONB))))
).first()
if obj is not None:
res.append(transRecord2DTO(session, permission))
where_str = transFilterTree(session, res, ds)
filters.append({"table": table.table_name, "filter": where_str})
if is_normal_user(current_user):
for table in table_list:
row_permissions = session.query(DsPermission).filter(
and_(DsPermission.table_id == table.id, DsPermission.type == 'row')).all()
contain_rules = session.query(DsRules).all()
res: List[PermissionDTO] = []
if row_permissions is not None:
for permission in row_permissions:
# check permission and user in same rules
# obj = session.query(DsRules).filter(
# and_(DsRules.permission_list.op('@>')(cast([permission.id], JSONB)),
# or_(DsRules.user_list.op('@>')(cast([f'{current_user.id}'], JSONB)),
# DsRules.user_list.op('@>')(cast([current_user.id], JSONB))))
# ).first()
flag = False
for r in contain_rules:
p_list = json.loads(r.permission_list)
u_list = json.loads(r.user_list)
if p_list is not None and u_list is not None and permission.id in p_list and (
current_user.id in u_list or f'{current_user.id}' in u_list):
flag = True
if flag:
res.append(transRecord2DTO(session, permission))
where_str = transFilterTree(session, res, ds)
filters.append({"table": table.table_name, "filter": where_str})
return filters
def get_column_permission_fields(session: SessionDep, current_user: CurrentUser, table: CoreTable, fields: list[CoreField]):
def get_column_permission_fields(session: SessionDep, current_user: CurrentUser, table: CoreTable,
fields: list[CoreField]):
if is_normal_user(current_user):
column_permissions = session.query(DsPermission).filter(
and_(DsPermission.table_id == table.id, DsPermission.type == 'column')).all()
contain_rules = session.query(DsRules).all()
if column_permissions is not None:
for permission in column_permissions:
# check permission and user in same rules
obj = session.query(DsRules).filter(
and_(DsRules.permission_list.op('@>')(cast([permission.id], JSONB)),
or_(DsRules.user_list.op('@>')(cast([f'{current_user.id}'], JSONB)),
DsRules.user_list.op('@>')(cast([current_user.id], JSONB))))
).first()
if obj is not None:
# obj = session.query(DsRules).filter(
# and_(DsRules.permission_list.op('@>')(cast([permission.id], JSONB)),
# or_(DsRules.user_list.op('@>')(cast([f'{current_user.id}'], JSONB)),
# DsRules.user_list.op('@>')(cast([current_user.id], JSONB))))
# ).first()
flag = False
for r in contain_rules:
p_list = json.loads(r.permission_list)
u_list = json.loads(r.user_list)
if p_list is not None and u_list is not None and permission.id in p_list and (
current_user.id in u_list or f'{current_user.id}' in u_list):
flag = True
if flag:
permission_list = json.loads(permission.permissions)
fields = filter_list(fields, permission_list)
return fields
def is_normal_user(current_user: CurrentUser):
return current_user.id != 1
def filter_list(list_a, list_b):
id_to_invalid = {}
for b in list_b:
if not b['enable']:
id_to_invalid[b['field_id']] = True
return [a for a in list_a if not id_to_invalid.get(a.id, False)]
return [a for a in list_a if not id_to_invalid.get(a.id, False)]
backend/common/core/config.py
-1
View File
@@ -85,4 +85,3 @@ class Settings(BaseSettings):
SERVER_EXCEL_HOST: str
settings = Settings() # type: ignore
print(settings)
backend/pyproject.toml
+1 -1
View File
@@ -36,7 +36,7 @@ dependencies = [
"pyyaml (>=6.0.2,<7.0.0)",
"fastapi-mcp (>=0.3.4,<0.4.0)",
"tabulate>=0.9.0",
"sqlbot-xpack==0.0.3.11",
"sqlbot-xpack==0.0.3.12",
"fastapi-cache2>=0.2.2",
"sqlparse>=0.5.3",
"redis>=6.2.0",