[24.0] Add require_default_history route argument

lib/galaxy/managers/session.py

@@ -6,18 +6,70 @@
 )
 from sqlalchemy.orm import joinedload
 
+from galaxy.model import (
+    GalaxySession,
+    History,
+)
 from galaxy.model.base import SharedModelMapping
+from galaxy.model.security import GalaxyRBACAgent
 
 log = logging.getLogger(__name__)
 
 
+def new_history(galaxy_session: GalaxySession, security_agent: GalaxyRBACAgent):
+    """
+    Create a new history and associate it with the current session and
+    its associated user (if set).
+    """
+    # Create new history
+    history = History()
+    # Associate with session
+    history.add_galaxy_session(galaxy_session)
+    # Make it the session's current history
+    galaxy_session.current_history = history
+    # Associate with user
+    if galaxy_session.user:
+        history.user = galaxy_session.user
+    # Set the user's default history permissions
+    security_agent.history_set_default_permissions(history)
+    return history
+
+
 class GalaxySessionManager:
     """Manages GalaxySession."""
 
     def __init__(self, model: SharedModelMapping):
         self.model = model
         self.sa_session = model.context
 
+    def get_or_create_default_history(self, galaxy_session: GalaxySession, security_agent: GalaxyRBACAgent):
+        default_history = galaxy_session.current_history
+        if default_history and not default_history.deleted:
+            return default_history
+        # history might be deleted, reset to None
+        default_history = None
+        user = galaxy_session.user
+        if user:
+            # Look for default history that (a) has default name + is not deleted and
+            # (b) has no datasets. If suitable history found, use it; otherwise, create
+            # new history.
+            stmt = select(History).filter_by(user=user, name=History.default_name, deleted=False)
+            unnamed_histories = self.sa_session.scalars(stmt)
+            for history in unnamed_histories:
+                if history.empty:
+                    # Found suitable default history.
+                    default_history = history
+                    break
+
+            # Set or create history.
+            if default_history:
+                galaxy_session.current_history = default_history
+        if not default_history:
+            default_history = new_history(galaxy_session, security_agent)
+            self.sa_session.add_all((default_history, galaxy_session))
+        self.sa_session.commit()
+        return default_history
+
     def get_session_from_session_key(self, session_key: str):
         """Returns GalaxySession if session_key is valid."""
         # going through self.model since this can be used by Galaxy or Toolshed despite

lib/galaxy/webapps/base/webapp.py

@@ -669,7 +669,7 @@ def _ensure_valid_session(self, session_cookie: str, create: bool = True) -> Non
         else:
             self.galaxy_session = galaxy_session
             if self.webapp.name == "galaxy":
-                self.get_or_create_default_history()
+                self.session_manager.get_or_create_default_history(galaxy_session, self.app.security_agent)
         # Do we need to flush the session?
         if galaxy_session_requires_flush:
             self.sa_session.add(galaxy_session)
@@ -908,7 +908,7 @@ def get_history(self, create=False, most_recent=False):
         if not history and most_recent:
             history = self.get_most_recent_history()
         if not history and util.string_as_bool(create):
-            history = self.get_or_create_default_history()
+            history = self.session_manager.get_or_create_default_history(self.galaxy_session, self.app.security_agent)
         return history
 
     def set_history(self, history):
@@ -922,40 +922,6 @@ def set_history(self, history):
     def history(self):
         return self.get_history()
 
-    def get_or_create_default_history(self):
-        """
-        Gets or creates a default history and associates it with the current
-        session.
-        """
-        history = self.galaxy_session.current_history
-        if history and not history.deleted:
-            return history
-
-        user = self.galaxy_session.user
-        if user:
-            # Look for default history that (a) has default name + is not deleted and
-            # (b) has no datasets. If suitable history found, use it; otherwise, create
-            # new history.
-            stmt = select(self.app.model.History).filter_by(
-                user=user, name=self.app.model.History.default_name, deleted=False
-            )
-            unnamed_histories = self.sa_session.scalars(stmt)
-            default_history = None
-            for history in unnamed_histories:
-                if history.empty:
-                    # Found suitable default history.
-                    default_history = history
-                    break
-
-            # Set or create history.
-            if default_history:
-                history = default_history
-                self.set_history(history)
-        else:
-            history = self.new_history()
-
-        return history
-
     def get_most_recent_history(self):
         """
         Gets the most recently updated history.

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

@@ -67,6 +67,7 @@
 )
 from galaxy.exceptions import (
     AdminRequiredException,
+    RequestParameterMissingException,
     UserCannotRunAsException,
 )
 from galaxy.managers.session import GalaxySessionManager
@@ -358,7 +359,19 @@ def get_admin_user(trans: SessionRequestContext = DependsOnTrans):
     return trans.user
 
 
+def get_or_create_default_history(
+    trans: SessionRequestContext = DependsOnTrans,
+    session_manager=cast(GalaxySessionManager, Depends(get_session_manager)),
+):
+    if not trans.user and not trans.galaxy_session:
+        raise RequestParameterMissingException("Cannot fetch or create history for unknown user session")
+    history = session_manager.get_or_create_default_history(trans.galaxy_session, trans.app.security_agent)
+    trans.history = history
+    return history
+
+
 AdminUserRequired = Depends(get_admin_user)
+DefaultHistoryRequired = Depends(get_or_create_default_history)
 
 
 class BaseGalaxyAPIController(BaseAPIController):
@@ -380,6 +393,7 @@ class FrameworkRouter(APIRouter):
     """A FastAPI Router tailored to Galaxy."""
 
     admin_user_dependency: Any
+    default_history_dependency: Any
 
     def wrap_with_alias(self, verb: RestVerb, *args, alias: Optional[str] = None, **kwd):
         """
@@ -463,6 +477,13 @@ def _handle_galaxy_kwd(self, kwd):
             else:
                 kwd["dependencies"] = [self.admin_user_dependency]
 
+        require_default_history = kwd.pop("require_default_history", False)
+        if require_default_history:
+            if "dependencies" in kwd:
+                kwd["dependencies"].append(self.default_history_dependency)
+            else:
+                kwd["dependencies"] = [self.default_history_dependency]
+
         public = kwd.pop("public", False)
         openapi_extra = kwd.pop("openapi_extra", {})
         if public:
@@ -483,6 +504,7 @@ def cbv(self):
 
 class Router(FrameworkRouter):
     admin_user_dependency = AdminUserRequired
+    default_history_dependency = DefaultHistoryRequired
 
 
 class APIContentTypeRoute(APIRoute):

lib/galaxy/webapps/galaxy/api/users.py

@@ -641,6 +641,7 @@ def index(
         "/api/users/{user_id}",
         name="get_user",
         summary="Return information about a specified or the current user. Only admin can see deleted or other users",
+        require_default_history=True,
     )
     def show(
         self,

lib/galaxy/work/context.py

@@ -70,6 +70,10 @@ def get_history(self, create=False):
     def history(self):
         return self.get_history()
 
+    @history.setter
+    def history(self, history):
+        self.__history = history
+
     def get_user(self):
         """Return the current user if logged in or None."""
         return self.__user

lib/galaxy_test/api/test_authenticate.py

@@ -75,3 +75,28 @@ def test_anon_history_creation(self):
             cookies=cookie,
         )
         assert second_histories_response.json()
+
+    def test_anon_history_creation_api(self):
+        # First request:
+        # We don't create any histories, just return a session cookie
+        response = get(self.url)
+        cookie = {"galaxysession": response.cookies["galaxysession"]}
+        # Check that we don't have any histories (API doesn't auto-create new histories)
+        histories_response = get(
+            urljoin(
+                self.url,
+                "api/histories",
+            )
+        )
+        assert not histories_response.json()
+        # Second request, we know client follows conventions by including cookies,
+        # default history is created.
+        get(urljoin(self.url, "api/users/current"), cookies=cookie).raise_for_status()
+        histories_response = get(
+            urljoin(
+                self.url,
+                "api/histories",
+            ),
+            cookies=cookie,
+        )
+        assert histories_response.json()

lib/tool_shed/webapp/buildapp.py

@@ -37,10 +37,6 @@ class ToolShedGalaxyWebTransaction(GalaxyWebTransaction):
     def repositories_hostname(self) -> str:
         return url_for("/", qualified=True).rstrip("/")
 
-    def get_or_create_default_history(self):
-        # tool shed has no concept of histories
-        raise NotImplementedError
-
 
 class CommunityWebApplication(galaxy.webapps.base.webapp.WebApplication):
     injection_aware: bool = True