Add helper method to determine if user has admin rights (#291)

Author: PGijsbers

src/core/access.py

@@ -2,7 +2,7 @@
 
 from sqlalchemy.engine import Row
 
-from database.users import User, UserGroup
+from database.users import User
 from schemas.datasets.openml import Visibility
 
 
@@ -17,5 +17,4 @@ async def _user_has_access(
         return False
     if user.user_id == dataset.uploader:
         return True
-    user_groups = await user.get_groups()
-    return UserGroup.ADMIN in user_groups
+    return await user.is_admin()

src/database/users.py

@@ -73,3 +73,6 @@ async def get_groups(self) -> list[UserGroup]:
             group_ids = await get_user_groups_for(user_id=self.user_id, connection=self._database)
             self._groups = [UserGroup(group_id) for group_id in group_ids]
         return self._groups
+
+    async def is_admin(self) -> bool:
+        return UserGroup.ADMIN in await self.get_groups()

src/routers/openml/datasets.py

@@ -32,7 +32,7 @@
     _format_dataset_url,
     _format_parquet_url,
 )
-from database.users import User, UserGroup
+from database.users import User
 from routers.dependencies import (
     Pagination,
     expdb_connection,
@@ -144,7 +144,7 @@ async def list_datasets(  # noqa: PLR0913, C901
 
     if user is None:
         clauses.append("AND `visibility`='public'")
-    elif UserGroup.ADMIN not in await user.get_groups():
+    elif not await user.is_admin():
         clauses.append("AND (`visibility`='public' OR `uploader`=:user_id)")
         parameters["user_id"] = user.user_id
 
@@ -347,12 +347,12 @@ async def update_dataset_status(
 
     dataset = await _get_dataset_raise_otherwise(dataset_id, user, expdb)
 
-    can_deactivate = dataset.uploader == user.user_id or UserGroup.ADMIN in await user.get_groups()
+    can_deactivate = dataset.uploader == user.user_id or await user.is_admin()
     if status == DatasetStatus.DEACTIVATED and not can_deactivate:
         msg = f"Dataset {dataset_id} is not owned by you."
         raise DatasetNotOwnedError(msg)
 
-    if status == DatasetStatus.ACTIVE and UserGroup.ADMIN not in await user.get_groups():
+    if status == DatasetStatus.ACTIVE and not await user.is_admin():
         msg = "Only administrators can activate datasets."
         raise DatasetAdminOnlyError(msg)
 

src/routers/openml/setups.py

@@ -13,7 +13,7 @@
     TagNotFoundError,
     TagNotOwnedError,
 )
-from database.users import User, UserGroup
+from database.users import User
 from routers.dependencies import expdb_connection, fetch_user_or_raise
 from routers.types import SystemString64
 
@@ -67,7 +67,7 @@ async def untag_setup(
         msg = f"Setup {setup_id} does not have tag {tag!r}."
         raise TagNotFoundError(msg)
 
-    if matched_tag_row.uploader != user.user_id and UserGroup.ADMIN not in await user.get_groups():
+    if matched_tag_row.uploader != user.user_id and not await user.is_admin():
         msg = (
             f"You may not remove tag {tag!r} of setup {setup_id} because it was not created by you."
         )

src/routers/openml/study.py

@@ -17,7 +17,7 @@
     StudyPrivateError,
 )
 from core.formatting import _str_to_bool
-from database.users import User, UserGroup
+from database.users import User
 from routers.dependencies import expdb_connection, fetch_user
 from schemas.core import Visibility
 from schemas.study import CreateStudy, Study, StudyStatus, StudyType
@@ -44,7 +44,7 @@ async def _get_study_raise_otherwise(
         if user is None:
             msg = "Must authenticate for private study."
             raise AuthenticationRequiredError(msg)
-        if study.creator != user.user_id and UserGroup.ADMIN not in await user.get_groups():
+        if study.creator != user.user_id and not await user.is_admin():
             msg = "Study is private."
             raise StudyPrivateError(msg)
     if _str_to_bool(study.legacy):
@@ -71,7 +71,7 @@ async def attach_to_study(
         raise AuthenticationRequiredError(msg)
     study = await _get_study_raise_otherwise(study_id, user, expdb)
     # PHP lets *anyone* edit *any* study. We're not going to do that.
-    if study.creator != user.user_id and UserGroup.ADMIN not in await user.get_groups():
+    if study.creator != user.user_id and not await user.is_admin():
         msg = f"Study {study_id} can only be edited by its creator."
         raise StudyNotEditableError(msg)
     if study.status != StudyStatus.IN_PREPARATION: