Clarify HTTP status codes for email and role authorization checks

Refactor API routers to use shared response definitions and update prefix

- Updated all API routers to use the new ROUTER_BASE_PREFIX for consistent URL structure.
- Replaced hardcoded response definitions with shared response mappings for better maintainability.
- Consolidated error responses into shared constants for easier updates and consistency across routers.
- Ensured that all routers now handle authentication and permission errors using the new shared response structure.

Refactor HTTP status codes for improved clarity and consistency across various endpoints

Update error handling for multiple sequences: change status code to 400 and provide explicit namespace guidance

Author: bencap

src/mavedb/lib/authorization.py

@@ -2,7 +2,6 @@
 from typing import Optional
 
 from fastapi import Depends, HTTPException
-from starlette import status
 
 from mavedb.lib.authentication import UserData, get_current_user
 from mavedb.lib.logging.context import logging_context, save_to_logging_context
@@ -21,10 +20,7 @@ async def require_current_user(
 ) -> UserData:
     if user_data is None:
         logger.info(msg="Non-authenticated user attempted to access protected route.", extra=logging_context())
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Could not validate credentials",
-        )
+        raise HTTPException(status_code=401, detail="Could not validate credentials")
 
     return user_data
 
@@ -38,8 +34,7 @@ async def require_current_user_with_email(
             msg="User attempted to access email protected route without a valid email.", extra=logging_context()
         )
         raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="There must be an email address associated with your account to use this feature.",
+            status_code=403, detail="There must be an email address associated with your account to use this feature."
         )
     return user_data
 
@@ -54,10 +49,7 @@ async def __call__(self, user_data: UserData = Depends(require_current_user)) ->
             logger.info(
                 msg="User attempted to access role protected route without a required role.", extra=logging_context()
             )
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="You are not authorized to use this feature",
-            )
+            raise HTTPException(status_code=403, detail="You are not authorized to use this feature")
 
         return user_data
 
@@ -68,9 +60,6 @@ async def require_role(roles: list[UserRole], user_data: UserData = Depends(requ
         logger.info(
             msg="User attempted to access role protected route without a required role.", extra=logging_context()
         )
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="You are not authorized to use this feature",
-        )
+        raise HTTPException(status_code=403, detail="You are not authorized to use this feature")
 
     return user_data

src/mavedb/lib/taxonomies.py

@@ -66,6 +66,6 @@ async def search_NCBI_taxonomy(db: Session, search: str) -> Any:
             else:
                 raise HTTPException(status_code=404, detail=f"Taxonomy with search {search_text} not found in NCBI")
     else:
-        raise HTTPException(status_code=404, detail="Please enter valid searching words")
+        raise HTTPException(status_code=400, detail="Search text is required")
 
     return taxonomy_record

src/mavedb/routers/access_keys.py

@@ -8,6 +8,7 @@
 from fastapi import APIRouter, Depends
 from fastapi.encoders import jsonable_encoder
 from fastapi.exceptions import HTTPException
+from sqlalchemy import and_
 from sqlalchemy.orm import Session
 
 from mavedb import deps
@@ -17,12 +18,13 @@
 from mavedb.lib.logging.context import logging_context, save_to_logging_context
 from mavedb.models.access_key import AccessKey
 from mavedb.models.enums.user_role import UserRole
+from mavedb.routers.shared import ACCESS_CONTROL_ERROR_RESPONSES, PUBLIC_ERROR_RESPONSES, ROUTER_BASE_PREFIX
 from mavedb.view_models import access_key
 
 router = APIRouter(
-    prefix="/api/v1",
+    prefix=f"{ROUTER_BASE_PREFIX}",
     tags=["Access Keys"],
-    responses={404: {"description": "Not found"}, 500: {"description": "Internal server error"}},
+    responses={**PUBLIC_ERROR_RESPONSES},
     route_class=LoggedRoute,
 )
 
@@ -49,9 +51,7 @@ def generate_key_pair():
     "/users/me/access-keys",
     status_code=200,
     response_model=list[access_key.AccessKey],
-    responses={
-        401: {"description": "Not authenticated"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
     summary="List my access keys",
 )
 def list_my_access_keys(*, user_data: UserData = Depends(require_current_user)) -> Any:
@@ -65,9 +65,7 @@ def list_my_access_keys(*, user_data: UserData = Depends(require_current_user))
     "/users/me/access-keys",
     status_code=200,
     response_model=access_key.NewAccessKey,
-    responses={
-        401: {"description": "Not authenticated"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
     summary="Create a new access key for myself",
 )
 def create_my_access_key(
@@ -94,10 +92,7 @@ def create_my_access_key(
     "/users/me/access-keys/{role}",
     status_code=200,
     response_model=access_key.NewAccessKey,
-    responses={
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
     summary="Create a new access key for myself with a specified role",
 )
 async def create_my_access_key_with_role(
@@ -138,9 +133,8 @@ async def create_my_access_key_with_role(
 @router.delete(
     "/users/me/access-keys/{key_id}",
     status_code=200,
-    responses={
-        401: {"description": "Not authenticated"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
+    summary="Delete one of my access keys",
 )
 def delete_my_access_key(
     *,
@@ -152,7 +146,9 @@ def delete_my_access_key(
     Delete one of the current user's access keys.
     """
     item = (
-        db.query(AccessKey).filter(AccessKey.key_id == key_id and AccessKey.user_id == user_data.user.id).one_or_none()
+        db.query(AccessKey)
+        .filter(and_(AccessKey.key_id == key_id, AccessKey.user_id == user_data.user.id))
+        .one_or_none()
     )
 
     if not item:

src/mavedb/routers/api_information.py

@@ -3,12 +3,13 @@
 from fastapi import APIRouter
 
 from mavedb import __project__, __version__
+from mavedb.routers.shared import PUBLIC_ERROR_RESPONSES, ROUTER_BASE_PREFIX
 from mavedb.view_models import api_version
 
 router = APIRouter(
-    prefix="/api/v1/api",
+    prefix=f"{ROUTER_BASE_PREFIX}/api",
     tags=["API Information"],
-    responses={404: {"description": "Not found"}, 500: {"description": "Internal server error"}},
+    responses={**PUBLIC_ERROR_RESPONSES},
 )
 
 

src/mavedb/routers/collections.py

@@ -24,14 +24,21 @@
 from mavedb.models.experiment import Experiment
 from mavedb.models.score_set import ScoreSet
 from mavedb.models.user import User
+from mavedb.routers.shared import (
+    ACCESS_CONTROL_ERROR_RESPONSES,
+    BASE_400_RESPONSE,
+    BASE_409_RESPONSE,
+    PUBLIC_ERROR_RESPONSES,
+    ROUTER_BASE_PREFIX,
+)
 from mavedb.view_models import collection, collection_bundle
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(
-    prefix="/api/v1",
+    prefix=f"{ROUTER_BASE_PREFIX}",
     tags=["Collections"],
-    responses={404: {"description": "Not found"}, 500: {"description": "Internal server error"}},
+    responses={**PUBLIC_ERROR_RESPONSES},
     route_class=LoggedRoute,
 )
 
@@ -41,9 +48,7 @@
     status_code=200,
     response_model=collection_bundle.CollectionBundle,
     response_model_exclude_none=True,
-    responses={
-        401: {"description": "Not authenticated"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
     summary="List my collections",
 )
 def list_my_collections(
@@ -96,10 +101,7 @@ def list_my_collections(
     "/collections/{urn}",
     status_code=200,
     response_model=collection.Collection,
-    responses={
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
     response_model_exclude_none=True,
     summary="Fetch a collection by URN",
 )
@@ -145,11 +147,7 @@ def fetch_collection(
 @router.post(
     "/collections/",
     response_model=collection.Collection,
-    responses={
-        400: {"description": "Bad request"},
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**BASE_400_RESPONSE, **ACCESS_CONTROL_ERROR_RESPONSES},
     response_model_exclude_none=True,
     summary="Create a collection",
 )
@@ -210,7 +208,7 @@ async def create_collection(
         save_to_logging_context(format_raised_exception_info_as_dict(e))
         logger.error(msg="Multiple users found with the given ORCID iD", extra=logging_context())
         raise HTTPException(
-            status_code=400,
+            status_code=500,
             detail="Multiple MaveDB users found with the given ORCID iD",
         )
 
@@ -233,7 +231,7 @@ async def create_collection(
     except MultipleResultsFound as e:
         save_to_logging_context(format_raised_exception_info_as_dict(e))
         logger.error(msg="Multiple resources found with the given URN", extra=logging_context())
-        raise HTTPException(status_code=400, detail="Multiple resources found with the given URN")
+        raise HTTPException(status_code=500, detail="Multiple resources found with the given URN")
 
     item = Collection(
         **jsonable_encoder(
@@ -266,11 +264,7 @@ async def create_collection(
 @router.patch(
     "/collections/{urn}",
     response_model=collection.Collection,
-    responses={
-        400: {"description": "Bad request"},
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**BASE_400_RESPONSE, **ACCESS_CONTROL_ERROR_RESPONSES},
     response_model_exclude_none=True,
     summary="Update a collection",
 )
@@ -421,11 +415,7 @@ async def add_score_set_to_collection(
 @router.delete(
     "/collections/{collection_urn}/score-sets/{score_set_urn}",
     response_model=collection.Collection,
-    responses={
-        400: {"description": "Bad request"},
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE},
     summary="Remove a score set from a collection",
 )
 async def delete_score_set_from_collection(
@@ -463,7 +453,7 @@ async def delete_score_set_from_collection(
             extra=logging_context(),
         )
         raise HTTPException(
-            status_code=400,
+            status_code=409,
             detail=f"association between score set '{score_set_urn}' and collection '{collection_urn}' not found",
         )
 
@@ -506,10 +496,7 @@ async def delete_score_set_from_collection(
 @router.post(
     "/collections/{collection_urn}/experiments",
     response_model=collection.Collection,
-    responses={
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
     summary="Add an experiment to a collection",
 )
 async def add_experiment_to_collection(
@@ -581,11 +568,7 @@ async def add_experiment_to_collection(
 @router.delete(
     "/collections/{collection_urn}/experiments/{experiment_urn}",
     response_model=collection.Collection,
-    responses={
-        400: {"description": "Bad request"},
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE},
     summary="Remove an experiment from a collection",
 )
 async def delete_experiment_from_collection(
@@ -623,7 +606,7 @@ async def delete_experiment_from_collection(
             extra=logging_context(),
         )
         raise HTTPException(
-            status_code=400,
+            status_code=409,
             detail=f"association between experiment '{experiment_urn}' and collection '{collection_urn}' not found",
         )
 
@@ -666,11 +649,7 @@ async def delete_experiment_from_collection(
 @router.post(
     "/collections/{urn}/{role}s",
     response_model=collection.Collection,
-    responses={
-        400: {"description": "Bad request"},
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE},
     summary="Add a user to a collection role",
 )
 async def add_user_to_collection_role(
@@ -723,7 +702,7 @@ async def add_user_to_collection_role(
             extra=logging_context(),
         )
         raise HTTPException(
-            status_code=400,
+            status_code=409,
             detail=f"user with ORCID iD '{body.orcid_id}' is already a {role} for collection '{urn}'",
         )
     # A user can only be in one role per collection, so remove from any other roles
@@ -757,11 +736,7 @@ async def add_user_to_collection_role(
 @router.delete(
     "/collections/{urn}/{role}s/{orcid_id}",
     response_model=collection.Collection,
-    responses={
-        400: {"description": "Bad request"},
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE},
     summary="Remove a user from a collection role",
 )
 async def remove_user_from_collection_role(
@@ -817,7 +792,7 @@ async def remove_user_from_collection_role(
             extra=logging_context(),
         )
         raise HTTPException(
-            status_code=400,
+            status_code=409,
             detail=f"user with ORCID iD '{orcid_id}' does not currently hold the role {role} for collection '{urn}'",
         )
 
@@ -845,10 +820,7 @@ async def remove_user_from_collection_role(
 
 @router.delete(
     "/collections/{urn}",
-    responses={
-        401: {"description": "Not authenticated"},
-        403: {"description": "User lacks necessary permissions"},
-    },
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
     summary="Delete a collection",
 )
 async def delete_collection(

src/mavedb/routers/controlled_keywords.py

@@ -6,12 +6,13 @@
 from mavedb import deps
 from mavedb.lib.keywords import search_keyword as _search_keyword
 from mavedb.models.controlled_keyword import ControlledKeyword
+from mavedb.routers.shared import PUBLIC_ERROR_RESPONSES, ROUTER_BASE_PREFIX
 from mavedb.view_models import keyword
 
 router = APIRouter(
-    prefix="/api/v1/controlled-keywords",
+    prefix=f"{ROUTER_BASE_PREFIX}/controlled-keywords",
     tags=["Controlled Keywords"],
-    responses={404: {"description": "Not found"}, 500: {"description": "Internal server error"}},
+    responses={**PUBLIC_ERROR_RESPONSES},
 )