Sync admin checking across old and new API frameworks.

lib/galaxy/web/framework/decorators.py

@@ -93,12 +93,7 @@ def require_admin(func):
     @wraps(func)
     def decorator(self, trans, *args, **kwargs):
         if not trans.user_is_admin:
-            msg = "You must be an administrator to access this feature."
-            user = trans.get_user()
-            if not trans.app.config.admin_users_list:
-                msg = "You must be logged in as an administrator to access this feature, but no administrators are set in the Galaxy configuration."
-            elif not user:
-                msg = "You must be logged in as an administrator to access this feature."
+            msg = require_admin_message(trans.app.config, trans.trans.get_user())
             trans.response.status = 403
             if trans.response.get_content_type() == 'application/json':
                 return msg
@@ -108,6 +103,16 @@ def decorator(self, trans, *args, **kwargs):
     return decorator
 
 
+def require_admin_message(config, user):
+    if not config.admin_users_list:
+        msg = "You must be logged in as an administrator to access this feature, but no administrators are set in the Galaxy configuration."
+    elif not user:
+        msg = "You must be logged in as an administrator to access this feature."
+    else:
+        msg = "You must be an administrator to access this feature."
+    return msg
+
+
 def do_not_cache(func):
     """
     Sets cache-prevention headers for the request.

lib/galaxy/webapps/galaxy/api/__init__.py

@@ -9,20 +9,20 @@
     Cookie,
     Depends,
     Header,
-    HTTPException,
     Query,
 )
 from sqlalchemy.orm import Session
 
 from galaxy import (
     app as galaxy_app,
-    exceptions,
     model,
 )
 from galaxy.app import UniverseApplication
+from galaxy.exceptions import AdminRequiredException
 from galaxy.managers.session import GalaxySessionManager
 from galaxy.managers.users import UserManager
 from galaxy.model import User
+from galaxy.web.framework.decorators import require_admin_message
 from galaxy.work.context import SessionRequestContext
 
 
@@ -58,10 +58,7 @@ def get_api_user(user_manager: UserManager = Depends(get_user_manager), key: Opt
     api_key = key or x_api_key
     if not api_key:
         return None
-    try:
-        return user_manager.by_api_key(api_key=api_key)
-    except exceptions.AuthenticationFailed as e:
-        raise HTTPException(status_code=e.status_code, detail=str(e))
+    return user_manager.by_api_key(api_key=api_key)
 
 
 def get_user(galaxy_session: Optional[model.GalaxySession] = Depends(get_session), api_user: Optional[User] = Depends(get_api_user)) -> Optional[User]:
@@ -78,7 +75,6 @@ def get_trans(app: UniverseApplication = Depends(get_app), user: Optional[User]
 
 
 def get_admin_user(trans: SessionRequestContext = Depends(get_trans), user_manager: UserManager = Depends(get_user_manager)):
-    if user_manager.is_admin(trans.user):
-        return trans.user
-    else:
-        raise HTTPException(status_code=403, detail="You must be an administrator to access this feature.")
+    if not trans.user_is_admin:
+        raise AdminRequiredException(require_admin_message(trans.app.config, trans.user))
+    return trans.user