feat: permission json to text

Author: XiaJunjie2020

backend/apps/datasource/crud/datasource.py

@@ -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

@@ -1,4 +1,3 @@
-
 import json
 from typing import List, Optional
 
@@ -11,56 +10,80 @@
 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

@@ -85,4 +85,3 @@ def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn | str:
     SERVER_EXCEL_HOST: str
 
 settings = Settings()  # type: ignore
-print(settings)

backend/pyproject.toml

@@ -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",