Merge pull request #14995 from mvdbeek/dev

Merge 22.05 into dev

lib/galaxy/managers/workflows.py

@@ -95,7 +95,10 @@ from galaxy.workflow.refactor.schema import (
 )
 from galaxy.workflow.reports import generate_report
 from galaxy.workflow.resources import get_resource_mapper_function
-from galaxy.workflow.steps import attach_ordered_steps
+from galaxy.workflow.steps import (
+    attach_ordered_steps,
+    has_cycles,
+)
 
 log = logging.getLogger(__name__)
 
@@ -863,7 +866,7 @@ class WorkflowContentsManager(UsesAnnotations):
         """
         if len(workflow.steps) == 0:
             raise exceptions.MessageException("Workflow cannot be run because it does not have any steps.")
-        if attach_ordered_steps(workflow):
+        if has_cycles(workflow):
             raise exceptions.MessageException("Workflow cannot be run because it contains cycles.")
         trans.workflow_building_mode = workflow_building_modes.USE_HISTORY
         module_injector = WorkflowModuleInjector(trans)
@@ -956,7 +959,7 @@ class WorkflowContentsManager(UsesAnnotations):
         """
         if len(workflow.steps) == 0:
             raise exceptions.MessageException("Workflow cannot be run because it does not have any steps.")
-        if attach_ordered_steps(workflow):
+        if has_cycles(workflow):
             raise exceptions.MessageException("Workflow cannot be run because it contains cycles.")
 
         # Ensure that the user has a history

lib/galaxy/tool_util/verify/interactor.py

@@ -833,6 +833,15 @@ class GalaxyInteractorApi:
         # no data for GET
         return requests.get(url, params=data, headers=headers, timeout=util.DEFAULT_SOCKET_TIMEOUT, **kwargs)
 
+    def _head(self, path, data=None, key=None, headers=None, admin=False, anon=False):
+        headers = self.api_key_header(key=key, admin=admin, anon=anon, headers=headers)
+        url = self.get_api_url(path)
+        kwargs: Dict[str, Any] = {}
+        if self.cookies:
+            kwargs["cookies"] = self.cookies
+        # no data for HEAD
+        return requests.head(url, params=data, headers=headers, timeout=util.DEFAULT_SOCKET_TIMEOUT, **kwargs)
+
     def get_api_url(self, path: str) -> str:
         if path.startswith("http"):
             return path

lib/galaxy/webapps/base/api.py

@@ -1,5 +1,11 @@
+import os
+import stat
+import typing
+
+import anyio
 from fastapi import (
     FastAPI,
+    HTTPException,
     Request,
     status,
 )
@@ -9,7 +15,10 @@ from starlette.middleware.base import (
     BaseHTTPMiddleware,
     RequestResponseEndpoint,
 )
-from starlette.responses import Response
+from starlette.responses import (
+    FileResponse,
+    Response,
+)
 from starlette_context.middleware import RawContextMiddleware
 from starlette_context.plugins import RequestIdPlugin
 
@@ -20,6 +29,132 @@ from galaxy.web.framework.decorators import (
     validation_error_to_message_exception,
 )
 
+if typing.TYPE_CHECKING:
+    from starlette.background import BackgroundTask
+    from starlette.types import (
+        Receive,
+        Scope,
+        Send,
+    )
+
+
+# Copied from https://github.com/tiangolo/fastapi/issues/1240#issuecomment-1055396884
+def _get_range_header(range_header: str, file_size: int) -> typing.Tuple[int, int]:
+    def _invalid_range():
+        return HTTPException(
+            status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
+            detail=f"Invalid request range (Range:{range_header!r})",
+        )
+
+    try:
+        h = range_header.replace("bytes=", "").rsplit("-", 1)
+        start = int(h[0]) if h[0] != "" else 0
+        end = int(h[1]) if h[1] != "" else file_size - 1
+    except ValueError:
+        raise _invalid_range()
+
+    if start > end or start < 0 or end > file_size - 1:
+        raise _invalid_range()
+    return start, end
+
+
+class GalaxyFileResponse(FileResponse):
+    """
+    Augments starlette FileResponse with x-accel-redirect/x-sendfile and byte-range handling.
+    """
+
+    nginx_x_accel_redirect_base: typing.Optional[str] = None
+    apache_xsendfile: typing.Optional[bool] = None
+    send_header_only: bool
+
+    def __init__(
+        self,
+        path: typing.Union[str, "os.PathLike[str]"],
+        status_code: int = 200,
+        headers: typing.Optional[typing.Mapping[str, str]] = None,
+        media_type: typing.Optional[str] = None,
+        background: typing.Optional["BackgroundTask"] = None,
+        filename: typing.Optional[str] = None,
+        stat_result: typing.Optional[os.stat_result] = None,
+        method: typing.Optional[str] = None,
+        content_disposition_type: str = "attachment",
+    ) -> None:
+        super().__init__(
+            path, status_code, headers, media_type, background, filename, stat_result, method, content_disposition_type
+        )
+        self.headers["accept-ranges"] = "bytes"
+        send_header_only = self.nginx_x_accel_redirect_base or self.apache_xsendfile
+        if self.nginx_x_accel_redirect_base:
+            self.headers["x-accel-redirect"] = self.nginx_x_accel_redirect_base + os.path.abspath(path)
+        elif self.apache_xsendfile:
+            self.headers["x-sendfile"] = os.path.abspath(path)
+        if not self.send_header_only and send_header_only:
+            # Not a head request, but nginx_x_accel_redirect_base / send_header_only, we don't send a body
+            self.send_header_only = True
+            self.headers["content-length"] = "0"
+
+    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
+        if self.stat_result is None:
+            try:
+                stat_result = await anyio.to_thread.run_sync(os.stat, self.path)
+                self.set_stat_headers(stat_result)
+            except FileNotFoundError:
+                raise RuntimeError(f"File at path {self.path} does not exist.")
+            else:
+                mode = stat_result.st_mode
+                if not stat.S_ISREG(mode):
+                    raise RuntimeError(f"File at path {self.path} is not a file.")
+
+        # This is where we diverge from the superclass, this adds support for byte range requests
+        start = 0
+        end = stat_result.st_size - 1
+        if not self.send_header_only:
+            http_range = ""
+            for key, value in scope["headers"]:
+                if key == b"range":
+                    http_range = value.decode("latin-1")
+                    start, end = _get_range_header(http_range, stat_result.st_size)
+                    self.headers["content-length"] = str(end - start + 1)
+                    self.headers["content-range"] = f"bytes {start}-{end}/{stat_result.st_size}"
+                    self.status_code = status.HTTP_206_PARTIAL_CONTENT
+                    break
+
+        await send(
+            {
+                "type": "http.response.start",
+                "status": self.status_code,
+                "headers": self.raw_headers,
+            }
+        )
+        if self.send_header_only:
+            await send({"type": "http.response.body", "body": b"", "more_body": False})
+        else:
+            # This also diverges from the superclass by seeking to start and limiting to end if handling byte range requests
+            async with await anyio.open_file(self.path, mode="rb") as file:
+                more_body = True
+                if start:
+                    await file.seek(start)
+                while more_body:
+                    if http_range:
+                        pos = await file.tell()
+                        read_size = min(self.chunk_size, end + 1 - pos)
+                        if pos + read_size == end + 1:
+                            more_body = False
+                    else:
+                        read_size = self.chunk_size
+                    chunk = await file.read(read_size)
+                    if more_body:
+                        more_body = len(chunk) == self.chunk_size
+                    await send(
+                        {
+                            "type": "http.response.body",
+                            "body": chunk,
+                            "more_body": more_body,
+                        }
+                    )
+        if self.background is not None:
+            await self.background()
+
 
 # Copied from https://stackoverflow.com/questions/71222144/runtimeerror-no-response-returned-in-fastapi-when-refresh-request/72677699#72677699
 class SuppressNoResponseReturnedMiddleware(BaseHTTPMiddleware):

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

@@ -269,6 +269,7 @@ class BaseGalaxyAPIController(BaseAPIController):
 
 class RestVerb(str, Enum):
     get = "GET"
+    head = "HEAD"
     post = "POST"
     put = "PUT"
     patch = "PATCH"
@@ -351,6 +352,9 @@ class Router(InferringRouter):
     def options(self, *args, **kwd):
         return self.wrap_with_alias(RestVerb.options, *args, **kwd)
 
+    def head(self, *args, **kwd):
+        return self.wrap_with_alias(RestVerb.head, *args, **kwd)
+
     def _handle_galaxy_kwd(self, kwd):
         require_admin = kwd.pop("require_admin", False)
         if require_admin:

lib/galaxy/webapps/galaxy/api/datasets.py

@@ -21,10 +21,7 @@ from fastapi import (
     Query,
     Request,
 )
-from starlette.responses import (
-    FileResponse,
-    StreamingResponse,
-)
+from starlette.responses import StreamingResponse
 
 from galaxy.schema import (
     FilterQueryParams,
@@ -40,6 +37,7 @@ from galaxy.schema.schema import (
     UpdateDatasetPermissionsPayload,
 )
 from galaxy.util.zipstream import ZipstreamWrapper
+from galaxy.webapps.base.api import GalaxyFileResponse
 from galaxy.webapps.galaxy.api import (
     depends,
     DependsOnTrans,
@@ -213,6 +211,16 @@ class FastAPIDatasets:
         tags=["histories"],
         response_class=StreamingResponse,
     )
+    @router.head(
+        "/api/datasets/{history_content_id}/display",
+        summary="Check if dataset content can be previewed or downloaded.",
+    )
+    @router.head(
+        "/api/histories/{history_id}/contents/{history_content_id}/display",
+        name="history_contents_display",
+        summary="Check if dataset content can be previewed or downloaded.",
+        tags=["histories"],
+    )
     def display(
         self,
         request: Request,
@@ -257,7 +265,7 @@ class FastAPIDatasets:
         if isinstance(display_data, IOBase):
             file_name = getattr(display_data, "name", None)
             if file_name:
-                return FileResponse(file_name, headers=headers)
+                return GalaxyFileResponse(file_name, headers=headers, method=request.method)
         elif isinstance(display_data, ZipstreamWrapper):
             return StreamingResponse(display_data.response(), headers=headers)
         elif isinstance(display_data, bytes):
@@ -268,12 +276,12 @@ class FastAPIDatasets:
         "/api/histories/{history_id}/contents/{history_content_id}/metadata_file",
         summary="Returns the metadata file associated with this history item.",
         tags=["histories"],
-        response_class=FileResponse,
+        response_class=GalaxyFileResponse,
     )
     @router.get(
         "/api/datasets/{history_content_id}/metadata_file",
         summary="Returns the metadata file associated with this history item.",
-        response_class=FileResponse,
+        response_class=GalaxyFileResponse,
     )
     def get_metadata_file(
         self,
@@ -289,7 +297,7 @@ class FastAPIDatasets:
         ),
     ):
         metadata_file_path, headers = self.service.get_metadata_file(trans, history_content_id, metadata_file)
-        return FileResponse(path=cast(str, metadata_file_path), headers=headers)
+        return GalaxyFileResponse(path=cast(str, metadata_file_path), headers=headers)
 
     @router.get(
         "/api/datasets/{dataset_id}",

lib/galaxy/webapps/galaxy/api/histories.py

@@ -21,7 +21,6 @@ from fastapi import (
 )
 from pydantic.fields import Field
 from pydantic.main import BaseModel
-from starlette.responses import FileResponse
 
 from galaxy.managers.context import (
     ProvidesHistoryContext,
@@ -51,6 +50,7 @@ from galaxy.schema.schema import (
     WriteStoreToPayload,
 )
 from galaxy.schema.types import LatestLiteral
+from galaxy.webapps.base.api import GalaxyFileResponse
 from galaxy.webapps.galaxy.api import (
     as_form,
     depends,
@@ -373,7 +373,7 @@ class FastAPIHistories:
         "/api/histories/{id}/exports/{jeha_id}",
         name="history_archive_download",
         summary=("If ready and available, return raw contents of exported history as a downloadable archive."),
-        response_class=FileResponse,
+        response_class=GalaxyFileResponse,
         responses={
             200: {
                 "description": "The archive file containing the History.",
@@ -394,7 +394,7 @@ class FastAPIHistories:
         jeha = self.service.get_ready_history_export(trans, id, jeha_id)
         media_type = self.service.get_archive_media_type(jeha)
         file_path = self.service.get_archive_download_path(trans, jeha)
-        return FileResponse(
+        return GalaxyFileResponse(
             path=file_path,
             media_type=media_type,
             filename=jeha.export_name,

lib/galaxy/webapps/galaxy/api/short_term_storage.py

@@ -1,13 +1,12 @@
 """
 API operations around galaxy.web.short_term_storage infrastructure.
 """
-from starlette.responses import FileResponse
-
 from galaxy.web.short_term_storage import (
     ShortTermStorageMonitor,
     ShortTermStorageServeCancelledInformation,
     ShortTermStorageServeCompletedInformation,
 )
+from galaxy.webapps.base.api import GalaxyFileResponse
 from . import (
     depends,
     Router,
@@ -34,7 +33,7 @@ class FastAPIShortTermStorage:
         "/api/short_term_storage/{storage_request_id}",
         summary="Serve the staged download specified by request ID.",
         response_description="Raw contents of the file.",
-        response_class=FileResponse,
+        response_class=GalaxyFileResponse,
         responses={
             200: {
                 "description": "The archive file containing the History.",
@@ -48,7 +47,7 @@ class FastAPIShortTermStorage:
         storage_target = self.short_term_storage_monitor.recover_target(storage_request_id)
         serve_info = self.short_term_storage_monitor.get_serve_info(storage_target)
         if isinstance(serve_info, ShortTermStorageServeCompletedInformation):
-            return FileResponse(
+            return GalaxyFileResponse(
                 path=serve_info.target.path,
                 media_type=serve_info.mime_type,
                 filename=serve_info.filename,

lib/galaxy/webapps/galaxy/api/tool_data.py

@@ -1,5 +1,4 @@
 from fastapi import Path
-from fastapi.responses import FileResponse
 
 from galaxy.managers.tool_data import ToolDataManager
 from galaxy.tool_util.data._schema import (
@@ -8,6 +7,7 @@ from galaxy.tool_util.data._schema import (
     ToolDataField,
     ToolDataItem,
 )
+from galaxy.webapps.base.api import GalaxyFileResponse
 from . import (
     depends,
     Router,
@@ -81,7 +81,7 @@ class FastAPIToolData:
         "/api/tool_data/{table_name}/fields/{field_name}/files/{file_name}",
         summary="Get information about a particular field in a tool data table",
         response_description="Information about a data table field",
-        response_class=FileResponse,
+        response_class=GalaxyFileResponse,
         require_admin=True,
     )
     async def download_field_file(
@@ -96,7 +96,7 @@ class FastAPIToolData:
     ):
         """Download a file associated with the data table field."""
         path = self.tool_data_manager.get_field_file_path(table_name, field_name, file_name)
-        return FileResponse(str(path))
+        return GalaxyFileResponse(str(path))
 
     @router.delete(
         "/api/tool_data/{table_name}",

lib/galaxy/webapps/galaxy/fast_app.py

@@ -1,22 +1,17 @@
-from pathlib import Path
-from typing import cast
-
 from a2wsgi import WSGIMiddleware
 from fastapi import (
     FastAPI,
     Request,
 )
 from starlette.middleware.cors import CORSMiddleware
-from starlette.responses import (
-    FileResponse,
-    Response,
-)
+from starlette.responses import Response
 
 from galaxy.version import VERSION
 from galaxy.webapps.base.api import (
     add_empty_response_middleware,
     add_exception_handler,
     add_request_id_middleware,
+    GalaxyFileResponse,
     include_all_package_routers,
 )
 from galaxy.webapps.base.webapp import config_allows_origin
@@ -105,29 +100,14 @@ def add_galaxy_middleware(app: FastAPI, gx_app):
             response.headers["X-Frame-Options"] = x_frame_options
             return response
 
-    nginx_x_accel_redirect_base = gx_app.config.nginx_x_accel_redirect_base
-    apache_xsendfile = gx_app.config.apache_xsendfile
+    GalaxyFileResponse.nginx_x_accel_redirect_base = gx_app.config.nginx_x_accel_redirect_base
+    GalaxyFileResponse.apache_xsendfile = gx_app.config.apache_xsendfile
 
     if gx_app.config.sentry_dsn:
         from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
 
         app.add_middleware(SentryAsgiMiddleware)
 
-    if nginx_x_accel_redirect_base or apache_xsendfile:
-
-        @app.middleware("http")
-        async def add_send_file_header(request: Request, call_next) -> Response:
-            response = await call_next(request)
-            if not isinstance(response, FileResponse):
-                return response
-            response = cast(FileResponse, response)
-            if nginx_x_accel_redirect_base:
-                full_path = Path(nginx_x_accel_redirect_base) / response.path
-                response.headers["X-Accel-Redirect"] = str(full_path)
-            if apache_xsendfile:
-                response.headers["X-Sendfile"] = str(response.path)
-            return response
-
     if gx_app.config.get("allowed_origin_hostnames", None):
         app.add_middleware(
             GalaxyCORSMiddleware,

lib/galaxy/workflow/steps.py

@@ -56,6 +56,14 @@ def order_workflow_steps(steps):
         return None
 
 
+def has_cycles(workflow):
+    try:
+        topsort(sorted(edgelist_for_workflow_steps(workflow.steps)))
+        return False
+    except CycleError:
+        return True
+
+
 def edgelist_for_workflow_steps(steps):
     """
     Create a list of tuples representing edges between ``WorkflowStep`` s based

lib/galaxy_test/api/test_datasets.py

@@ -321,12 +321,42 @@ class TestDatasetsApi(ApiTestCase):
         10  20  30  40
         """
         )
-        hda1 = self.dataset_populator.new_dataset(history_id, content=contents)
-        self.dataset_populator.wait_for_history(history_id)
+        hda1 = self.dataset_populator.new_dataset(history_id, content=contents, wait=True)
         display_response = self._get(f"histories/{history_id}/contents/{hda1['id']}/display", {"raw": "True"})
         self._assert_status_code_is(display_response, 200)
         assert display_response.text == contents
 
+    def test_head(self, history_id):
+        hda1 = self.dataset_populator.new_dataset(history_id, wait=True)
+        display_response = self._head(f"histories/{history_id}/contents/{hda1['id']}/display", {"raw": "True"})
+        self._assert_status_code_is(display_response, 200)
+        assert display_response.text == ""
+        display_response = self._head(
+            f"histories/{history_id}/contents/{hda1['id']}{hda1['id']}/display", {"raw": "True"}
+        )
+        self._assert_status_code_is(display_response, 400)
+
+    def test_byte_range_support(self, history_id):
+        hda1 = self.dataset_populator.new_dataset(history_id, wait=True)
+        head_response = self._head(f"histories/{history_id}/contents/{hda1['id']}/display", {"raw": "True"})
+        self._assert_status_code_is(head_response, 200)
+        assert head_response.headers["content-length"] == "12"
+        assert head_response.text == ""
+        assert head_response.headers["accept-ranges"] == "bytes"
+        valid_headers = {"range": "bytes=0-0"}
+        display_response = self._get(
+            f"histories/{history_id}/contents/{hda1['id']}/display", {"raw": "True"}, headers=valid_headers
+        )
+        self._assert_status_code_is(display_response, 206)
+        assert len(display_response.text) == 1
+        assert display_response.headers["content-length"] == "1"
+        assert display_response.headers["content-range"] == "bytes 0-0/12"
+        invalid_headers = {"range": "bytes=-1-1"}
+        display_response = self._get(
+            f"histories/{history_id}/contents/{hda1['id']}/display", {"raw": "True"}, headers=invalid_headers
+        )
+        self._assert_status_code_is(display_response, 416)
+
     def test_tag_change(self, history_id):
         hda_id = self.dataset_populator.new_dataset(history_id)["id"]
         payload = {

lib/galaxy_test/base/api.py

@@ -148,6 +148,9 @@ class UsesApiTestCaseMixin:
     def _get(self, *args, **kwds):
         return self.galaxy_interactor.get(*args, **kwds)
 
+    def _head(self, *args, **kwds):
+        return self.galaxy_interactor.head(*args, **kwds)
+
     def _post(self, *args, **kwds):
         return self.galaxy_interactor.post(*args, **kwds)
 
@@ -199,6 +202,9 @@ class ApiTestInteractor(BaseInteractor):
     def get(self, *args, **kwds):
         return self._get(*args, **kwds)
 
+    def head(self, *args, **kwds):
+        return self._head(*args, **kwds)
+
     def post(self, *args, **kwds):
         return self._post(*args, **kwds)
 

test/integration/test_apache_nginx_sendfile.py

@@ -0,0 +1,49 @@
+import os
+
+from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.driver.integration_util import IntegrationTestCase
+
+
+class TestNginxAccelHeader(IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["nginx_x_accel_redirect_base"] = "/redirect"
+
+    def test_dataset_download(self, history_id):
+        hda = self.dataset_populator.new_dataset(history_id=history_id, wait=True)
+        head_response = self._head(f"histories/{history_id}/contents/{hda['id']}/display", {"raw": "True"})
+        self._assert_status_code_is(head_response, 200)
+        assert head_response.headers["content-length"] == "12"
+        display_response = self._get(f"histories/{history_id}/contents/{hda['id']}/display", {"raw": "True"})
+        self._assert_status_code_is(display_response, 200)
+        assert display_response.headers["content-length"] == "0"
+        assert display_response.headers["x-accel-redirect"].startswith("/redirect")
+
+
+class TestApacheSendFileHeader(IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["apache_xsendfile"] = True
+
+    def test_dataset_download(self, history_id):
+        hda = self.dataset_populator.new_dataset(history_id=history_id, wait=True)
+        head_response = self._head(f"histories/{history_id}/contents/{hda['id']}/display", {"raw": "True"})
+        self._assert_status_code_is(head_response, 200)
+        assert head_response.headers["content-length"] == "12"
+        display_response = self._get(f"histories/{history_id}/contents/{hda['id']}/display", {"raw": "True"})
+        self._assert_status_code_is(display_response, 200)
+        assert display_response.headers["content-length"] == "0"
+        assert "x-sendfile" in display_response.headers
+        assert os.path.exists(display_response.headers["x-sendfile"])