Modify original check authorization function. Based on this new funct…

…ion, add more tests and move original tests to one file.

src/mavedb/routers/authorization.py

@@ -0,0 +1,71 @@
+import logging
+from enum import Enum
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from mavedb import deps
+from mavedb.lib.authentication import get_current_user, UserData
+from mavedb.lib.permissions import has_permission, Action
+from mavedb.lib.logging import LoggedRoute
+from mavedb.lib.logging.context import logging_context, save_to_logging_context
+from mavedb.models.experiment import Experiment
+from mavedb.models.experiment_set import ExperimentSet
+from mavedb.models.score_set import ScoreSet
+
+router = APIRouter(
+    prefix="/api/v1",
+    tags=["authorizations"],
+    responses={404: {"description": "Not found"}},
+    route_class=LoggedRoute,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ModelName(str, Enum):
+    experiment = "experiment"
+    experiment_set = "experiment-set"
+    score_set = "score-set"
+
+
+@router.get(
+    "/user-is-authorized/{model_name}/{urn}/{action}",
+    status_code=200,
+    response_model=bool
+)
+async def check_authorization(
+    *,
+    model_name: str,
+    urn: str,
+    action: str,
+    db: Session = Depends(deps.get_db),
+    user_data: UserData = Depends(get_current_user),
+) -> bool:
+    """
+    Check whether users have authorizations in adding/editing/deleting/publishing experiment or score set.
+    """
+    save_to_logging_context({"requested_resource": urn})
+    if model_name == ModelName.experiment_set:
+        item = db.query(ExperimentSet).filter(ExperimentSet.urn == urn).one_or_none()
+    elif model_name == ModelName.experiment:
+        item = db.query(Experiment).filter(Experiment.urn == urn).one_or_none()
+    elif model_name == ModelName.score_set:
+        item = db.query(ScoreSet).filter(ScoreSet.urn == urn).one_or_none()
+    else:
+        item = None
+
+    if item:
+        if user_data:
+            try:
+                action_enum = Action[action.upper()]
+                permission = has_permission(user_data, item, action_enum).permitted
+                return permission
+            except KeyError:
+                raise HTTPException(status_code=400, detail=f"Invalid action: {action}")
+        else:
+            logger.debug(msg="Miss user data", extra=logging_context())
+            raise HTTPException(status_code=404, detail=f"User not found")
+    else:
+        logger.debug(msg="The requested resources does not exist.", extra=logging_context())
+        raise HTTPException(status_code=404, detail=f"{model_name} with URN '{urn}' not found")

src/mavedb/routers/experiment_sets.py

@@ -4,14 +4,12 @@
 
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
-from sqlalchemy import or_
 
 from mavedb import deps
 from mavedb.lib.authentication import get_current_user, UserData
 from mavedb.lib.permissions import has_permission, Action
 from mavedb.lib.logging import LoggedRoute
 from mavedb.lib.logging.context import logging_context, save_to_logging_context
-from mavedb.models.contributor import Contributor
 from mavedb.models.experiment_set import ExperimentSet
 from mavedb.view_models import experiment_set
 
@@ -25,40 +23,6 @@
 logger = logging.getLogger(__name__)
 
 
-@router.get(
-    "/check-authorizations/{urn}",
-    status_code=200,
-    response_model=bool
-)
-async def check_experiment_set_authorization(
-    *,
-    urn: str,
-    db: Session = Depends(deps.get_db),
-    user_data: UserData = Depends(get_current_user),
-) -> bool:
-    """
-    Check whether users have authorizations in this experiment set.
-    """
-    query = db.query(ExperimentSet).filter(ExperimentSet.urn == urn)
-
-    if user_data is not None:
-        query = query.filter(
-            or_(
-                ExperimentSet.created_by_id == user_data.user.id,
-                ExperimentSet.contributors.any(Contributor.orcid_id == user_data.user.username),
-            )
-        )
-    else:
-        return False
-
-    save_to_logging_context({"Experiment set requested resource": urn})
-    item = query.first()
-    if item:
-        return True
-    else:
-        return False
-
-
 @router.get(
     "/{urn}",
     status_code=200,

src/mavedb/routers/experiments.py

@@ -5,7 +5,6 @@
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.encoders import jsonable_encoder
 import pydantic
-from sqlalchemy import or_, and_
 from sqlalchemy.orm import Session
 
 from mavedb import deps
@@ -45,41 +44,6 @@
 )
 
 
-@router.get(
-    "/experiments/check-authorizations/{urn}",
-    status_code=200,
-    response_model=bool
-)
-async def check_experiment_authorization(
-    *,
-    urn: str,
-    db: Session = Depends(deps.get_db),
-    user_data: UserData = Depends(get_current_user),
-) -> bool:
-    """
-    Check whether users have authorizations in this experiment.
-    """
-    query = db.query(Experiment).filter(Experiment.urn == urn)
-
-    if user_data is not None:
-        query = query.filter(
-            or_(
-                Experiment.created_by_id == user_data.user.id,
-                Experiment.contributors.any(Contributor.orcid_id == user_data.user.username),
-            )
-        )
-    else:
-        return False
-
-    save_to_logging_context({"Experiment requested resource": urn})
-    item = query.first()
-
-    if item:
-        return True
-    else:
-        return False
-
-
 # TODO: Rewrite this function.
 @router.get(
     "/experiments/",

src/mavedb/routers/score_sets.py

@@ -113,40 +113,6 @@ async def fetch_score_set_by_urn(
 )
 
 
-@router.get(
-    "/score-sets/check-authorizations/{urn}",
-    status_code=200,
-    response_model=bool
-)
-async def check_score_set_authorization(
-    *,
-    urn: str,
-    db: Session = Depends(deps.get_db),
-    user_data: UserData = Depends(get_current_user),
-) -> bool:
-    """
-    Check whether users have authorizations in this score set.
-    """
-    query = db.query(ScoreSet).filter(ScoreSet.urn == urn)
-
-    if user_data is not None:
-        query = query.filter(
-            or_(
-                ScoreSet.created_by_id == user_data.user.id,
-                ScoreSet.contributors.any(Contributor.orcid_id == user_data.user.username),
-            )
-        )
-    else:
-        return False
-
-    save_to_logging_context({"Score set requested resource": urn})
-    item = query.first()
-    if item:
-        return True
-    else:
-        return False
-
-
 @router.post("/score-sets/search", status_code=200, response_model=list[score_set.ShortScoreSet])
 def search_score_sets(search: ScoreSetsSearch, db: Session = Depends(deps.get_db)) -> Any:  # = Body(..., embed=True),
     """

src/mavedb/server_main.py

@@ -26,6 +26,7 @@
 from mavedb.lib.logging.canonical import log_request
 from mavedb.routers import (
     access_keys,
+    authorization,
     api_information,
     controlled_keywords,
     doi_identifiers,
@@ -73,6 +74,7 @@
 )
 app.include_router(access_keys.router)
 app.include_router(api_information.router)
+app.include_router(authorization.router)
 app.include_router(controlled_keywords.router)
 app.include_router(doi_identifiers.router)
 app.include_router(experiment_sets.router)