patch group logic and filtering

Author: petrCher

auth_backend/admin/admin.py

@@ -3,10 +3,11 @@
 from sqlalchemy.sql.expression import Select
 from starlette.requests import Request
 
+from auth_backend.admin.filter import FilteredModelConverter
 from auth_backend.models.db import Group, Scope, User
-from auth_backend.routes.groups import create_group_logic, delete_group_id
+from auth_backend.routes.groups import create_group_logic, delete_group_id, patch_group_logic
 from auth_backend.routes.user import patch_user_groups
-from auth_backend.schemas.models import GroupPost
+from auth_backend.schemas.models import GroupPatch, GroupPost
 
 
 class ScopeAdmin(ModelView, model=Scope):
@@ -27,6 +28,7 @@ class ScopeAdmin(ModelView, model=Scope):
     column_default_sort = [("id", False)]
     form_excluded_columns = ["create_ts", "update_ts", "groups", "user_sessions", "is_deleted"]
     can_create = False  # I don't know how to use UnionAuth there to get user_id that is required
+    form_converter = FilteredModelConverter
 
     def list_query(self, request: Request) -> Select:
         return select(Scope).where(Scope.is_deleted == False)
@@ -65,14 +67,14 @@ class GroupAdmin(ModelView, model=Group):
     column_sortable_list = ["id", "name", "parent_id", "is_deleted"]
     column_default_sort = [("id", False)]
     form_excluded_columns = ["create_ts", "update_ts", "is_deleted"]
+    form_converter = FilteredModelConverter
 
     def list_query(self, request: Request) -> Select:
         return select(Group).where(Group.is_deleted == False)
 
     def count_query(self, request: Request) -> Select:
         return select(func.count(Group.id)).where(Group.is_deleted == False)
 
-    # add logic with deleted scopes showed
     async def insert_model(self, request, data):
         scope_ids = [int(s) for s in (data.pop("scopes", None) or [])]
         parent_id = int(data["parent_id"]) if data.get("parent_id") else None
@@ -81,7 +83,17 @@ async def insert_model(self, request, data):
             result = create_group_logic(group_inp, session)
             return Group.get(result["id"], session=session)
 
-    # add update group
+    async def update_model(self, request, pk, data):
+        scope_ids = [int(s) for s in (data.pop("scopes", None) or [])]
+        parent_id = int(data["parent_id"]) if data.get("parent_id") else None
+        group_inp = GroupPatch(
+            name=data.get("name"),
+            parent_id=parent_id,
+            scopes=scope_ids,
+        )
+        with self.session_maker(expire_on_commit=False) as session:
+            return patch_group_logic(int(pk), group_inp, session)
+
     async def delete_model(self, request, pk):
         with self.session_maker(expire_on_commit=False) as session:
             delete_group_id(int(pk), session)
@@ -104,6 +116,7 @@ class UserAdmin(ModelView, model=User):
     column_formatters_detail = {
         "scopes": lambda m, a: ", ".join(s.name for s in (m.scopes or set())),
     }
+    form_converter = FilteredModelConverter
 
     def list_query(self, request: Request) -> Select:
         return select(User).where(User.is_deleted == False)

auth_backend/admin/filter.py

@@ -0,0 +1,20 @@
+import anyio
+from sqladmin.forms import ModelConverter
+from sqladmin.helpers import is_async_session_maker
+from sqlalchemy import select
+
+
+class FilteredModelConverter(ModelConverter):
+    async def _prepare_select_options(self, prop, session_maker):
+        target_model = prop.mapper.class_
+        stmt = select(target_model)
+        if hasattr(target_model, "is_deleted"):
+            stmt = stmt.where(target_model.is_deleted == False)
+        if is_async_session_maker(session_maker):
+            async with session_maker() as session:
+                objects = await session.execute(stmt)
+                return [(str(self._get_identifier_value(obj)), str(obj)) for obj in objects.scalars().unique().all()]
+        else:
+            with session_maker() as session:
+                objects = await anyio.to_thread.run_sync(session.execute, stmt)
+                return [(str(self._get_identifier_value(obj)), str(obj)) for obj in objects.scalars().unique().all()]

auth_backend/routes/groups.py

@@ -73,47 +73,52 @@ async def create_group(
     return Group(**result).model_dump(exclude_unset=True)
 
 
-@groups.patch("/{id}", response_model=Group)
-async def patch_group(
-    id: int,
-    group_inp: GroupPatch,
-    _: UserSession = Depends(UnionAuth(scopes=["auth.group.update"], allow_none=False, auto_error=True)),
-) -> Group:
-    """
-    Scopes: `["auth.group.update"]`
-    """
+def patch_group_logic(id: int, group_inp: GroupPatch, session) -> DbGroup:
     if (
-        exists_check := DbGroup.query(session=db.session)
+        exists_check := DbGroup.query(session=session)
         .filter(DbGroup.name == group_inp.name, DbGroup.id != id)
         .one_or_none()
     ):
         raise AlreadyExists(Group, exists_check.id)
-    group = DbGroup.get(id, session=db.session)
+    group = DbGroup.get(id, session=session)
     if group_inp.parent_id in (row.id for row in group.child):
         raise HTTPException(
             status_code=400,
             detail=StatusResponseModel(status="Error", message="Cycle detected", ru="Найден цикл").model_dump(),
         )
     result = Group.model_validate(
-        DbGroup.update(id, session=db.session, **group_inp.model_dump(exclude_unset=True, exclude={"scopes"}))
+        DbGroup.update(id, session=session, **group_inp.model_dump(exclude_unset=True, exclude={"scopes"}))
     ).model_dump(exclude_unset=True)
     scopes = set()
     if group_inp.scopes is not None:
         for _scope_id in group_inp.scopes:
-            scopes.add(Scope.get(session=db.session, id=_scope_id))
+            scopes.add(Scope.get(session=session, id=_scope_id))
         group.scopes = scopes
-    db.session.commit()
+    session.commit()
+    return group
+
+
+@groups.patch("/{id}", response_model=Group)
+async def patch_group(
+    id: int,
+    group_inp: GroupPatch,
+    _: UserSession = Depends(UnionAuth(scopes=["auth.group.update"], allow_none=False, auto_error=True)),
+) -> Group:
+    """
+    Scopes: `["auth.group.update"]`
+    """
+    group = patch_group_logic(id, group_inp, db.session)
     return Group.model_validate(group)
 
 
 def delete_group_id(id: int, session) -> None:
-    group: DbGroup = DbGroup.get(id, session=db.session)
+    group: DbGroup = DbGroup.get(id, session=session)
     if child := group.child:
         for children in child:
             children.parent_id = group.parent_id
-        db.session.flush()
-    DbGroup.delete(id, session=db.session)
-    db.session.commit()
+        session.flush()
+    DbGroup.delete(id, session=session)
+    session.commit()
 
 
 @groups.delete("/{id}", response_model=None)