Is1865/fix download outputs (#2275)

Outputs produced by a project cannot be retrieved as a file resource in the public-api because their naming is incompatible. This PR overcomes this problem by creating a soft link to the output files with a name compatible with the public api

*api-server
ADDED: creates soft links to output files named api/{file_id}/{file_name}
CHANGED: extends testing

*storage
ADDED: new entry to create soft links from existing file POST /files/{file_id}:soft-copy
CHANGED: new column file_meta_data.is_soft_link: bool
CHANGED: refactors routing
CHANGED: extends testing

*simcore-sdk
CHANGED: improved test fixture store_link to upload properly, i.e. via storage service

Author: pcrespov

api/specs/storage/openapi.yaml

@@ -417,6 +417,40 @@ paths:
         "204":
           description: folder has been successfully deleted
 
+  /files/{file_id}:soft-copy:
+    post:
+      summary: Copy as soft link
+      operationId: copy_as_soft_link
+      parameters:
+        - name: file_id
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: user_id
+          in: query
+          required: true
+          schema:
+            type: integer
+      requestBody:
+        content:
+          application/json:
+            schema:
+              type: object
+              required:
+                - link_id
+              properties:
+                link_id:
+                  type: string
+      responses:
+        "200":
+          description: "Returns link metadata"
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/FileMetaDataEnveloped"
+        default:
+          $ref: "#/components/responses/DefaultErrorResponse"
 components:
   schemas:
     HealthCheckEnveloped:

packages/models-library/src/models_library/api_schemas_storage.py

@@ -153,6 +153,11 @@ class FileMetaData(BaseModel):
     entity_tag: Optional[str] = Field(
         description="Entity tag (or ETag), represents a specific version of the file",
     )
+    is_soft_link: bool = Field(
+        False,
+        description="If true, this file is a soft link."
+        "i.e. is another entry with the same object_name",
+    )
 
     parent_id: Optional[str]
 

packages/postgres-database/src/simcore_postgres_database/migration/versions/be29dbed0cce_adds_file_meta_data_is_soft_link_column.py

@@ -0,0 +1,28 @@
+"""adds file_meta_data.is_soft_link column
+
+Revision ID: be29dbed0cce
+Revises: e43bd59a8e17
+Create Date: 2021-04-15 08:10:50.878539+00:00
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'be29dbed0cce'
+down_revision = 'e43bd59a8e17'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('file_meta_data', sa.Column('is_soft_link', sa.Boolean(), server_default=sa.text('false'), nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('file_meta_data', 'is_soft_link')
+    # ### end Alembic commands ###

packages/postgres-database/src/simcore_postgres_database/models/file_meta_data.py

@@ -5,25 +5,37 @@
 file_meta_data = sa.Table(
     "file_meta_data",
     metadata,
-    sa.Column("file_uuid", sa.String, primary_key=True),
-    sa.Column("location_id", sa.String),
-    sa.Column("location", sa.String),
-    sa.Column("bucket_name", sa.String),
-    sa.Column("object_name", sa.String),
-    sa.Column("project_id", sa.String),
-    sa.Column("project_name", sa.String),
-    sa.Column("node_id", sa.String),
-    sa.Column("node_name", sa.String),
-    sa.Column("file_name", sa.String),
-    sa.Column("user_id", sa.String),
-    sa.Column("user_name", sa.String),
-    sa.Column("file_id", sa.String),
-    sa.Column("raw_file_path", sa.String),
-    sa.Column("display_file_path", sa.String),
-    sa.Column("created_at", sa.String),
-    sa.Column("last_modified", sa.String),
-    sa.Column("file_size", sa.BigInteger),
-    # Entity tag (or ETag), represents a specific version of the object.
-    # SEE https://docs.aws.amazon.com/AmazonS3/latest/API/RESTCommonResponseHeaders.html)
-    sa.Column("entity_tag", sa.String, nullable=True),
+    sa.Column("file_uuid", sa.String(), primary_key=True),
+    sa.Column("location_id", sa.String()),
+    sa.Column("location", sa.String()),
+    sa.Column("bucket_name", sa.String()),
+    sa.Column("object_name", sa.String()),
+    sa.Column("project_id", sa.String()),
+    sa.Column("project_name", sa.String()),
+    sa.Column("node_id", sa.String()),
+    sa.Column("node_name", sa.String()),
+    sa.Column("file_name", sa.String()),
+    sa.Column("user_id", sa.String()),
+    sa.Column("user_name", sa.String()),
+    sa.Column("file_id", sa.String()),
+    sa.Column("raw_file_path", sa.String()),
+    sa.Column("display_file_path", sa.String()),
+    sa.Column("created_at", sa.String()),
+    sa.Column("last_modified", sa.String()),
+    sa.Column("file_size", sa.BigInteger()),
+    sa.Column(
+        "entity_tag",
+        sa.String(),
+        nullable=True,
+        doc="Entity tag (or ETag), represents a specific version of the object"
+        "SEE https://docs.aws.amazon.com/AmazonS3/latest/API/RESTCommonResponseHeaders.html",
+    ),
+    sa.Column(
+        "is_soft_link",
+        sa.Boolean(),
+        nullable=False,
+        server_default=sa.text("false"),
+        doc="If true, this file is a soft link."
+        "i.e. is another entry with the same object_name",
+    ),
 )

packages/pytest-simcore/src/pytest_simcore/minio_service.py

@@ -74,6 +74,7 @@ def wait_till_minio_responsive(minio_config: Dict[str, str]) -> bool:
 def bucket(minio_config: Dict[str, str], minio_service: S3Client) -> str:
     bucket_name = minio_config["bucket_name"]
     minio_service.create_bucket(bucket_name, delete_contents_if_exists=True)
+
     yield bucket_name
 
     minio_service.remove_bucket(bucket_name, delete_contents=True)

packages/pytest-simcore/src/pytest_simcore/simcore_services.py

@@ -63,10 +63,7 @@ async def simcore_services(services_endpoint: Dict[str, URL], monkeypatch) -> No
 async def wait_till_service_responsive(endpoint: URL):
     async with aiohttp.ClientSession() as session:
         async with session.get(endpoint) as resp:
+            # NOTE: Health-check endpoint require only a
+            # status code 200 (see e.g. services/web/server/docker/healthcheck.py)
+            # regardless of the payload content
             assert resp.status == 200
-            data = await resp.json()
-            # aiohttp based services are like this:
-            assert "data" in data or ":-)" in data or ":-)" in data.get("msg")
-            if "data" in data:
-                assert "status" in data["data"]
-                assert data["data"]["status"] == "SERVICE_RUNNING"

packages/pytest-simcore/src/pytest_simcore/simcore_storage_service.py

@@ -50,5 +50,4 @@ async def wait_till_storage_responsive(storage_endpoint: URL):
             assert resp.status == 200
             data = await resp.json()
             assert "data" in data
-            assert "status" in data["data"]
-            assert data["data"]["status"] == "SERVICE_RUNNING"
+            assert data["data"] is not None

packages/pytest-simcore/src/pytest_simcore/simcore_webserver_service.py

@@ -7,9 +7,8 @@
 import aiohttp
 import pytest
 import tenacity
-from yarl import URL
-
 from servicelib.minio_utils import MinioRetryPolicyUponInitialization
+from yarl import URL
 
 from .helpers.utils_docker import get_service_published_port
 
@@ -36,8 +35,7 @@ async def webserver_service(webserver_endpoint: URL, docker_stack: Dict) -> URL:
 async def wait_till_webserver_responsive(webserver_endpoint: URL):
     async with aiohttp.ClientSession() as session:
         async with session.get(webserver_endpoint.with_path("/v0/")) as resp:
+            # NOTE: Health-check endpoint require only a
+            # status code 200 (see e.g. services/web/server/docker/healthcheck.py)
+            # regardless of the payload content
             assert resp.status == 200
-            data = await resp.json()
-            assert "data" in data
-            assert "status" in data["data"]
-            assert data["data"]["status"] == "SERVICE_RUNNING"

packages/simcore-sdk/src/simcore_sdk/node_ports/filemanager.py

@@ -100,35 +100,40 @@ async def _get_location_id_from_location_name(store: str, api: UsersApi):
 
 async def _get_link(store_id: int, file_id: str, apifct) -> URL:
     log.debug("Getting link from store id %s for %s", store_id, file_id)
-    try:
-        # When uploading and downloading files from the storage service
-        # it is important to use a longer timeout, previously was 5 minutes
-        # changing to 1 hour. this will allow for larger payloads to be stored/download
-        resp = await apifct(
-            location_id=store_id,
-            user_id=config.USER_ID,
-            file_id=file_id,
-            _request_timeout=ServicesCommonSettings().storage_service_upload_download_timeout,
-        )
+    # When uploading and downloading files from the storage service
+    # it is important to use a longer timeout, previously was 5 minutes
+    # changing to 1 hour. this will allow for larger payloads to be stored/download
+    resp = await apifct(
+        location_id=store_id,
+        user_id=config.USER_ID,
+        file_id=file_id,
+        _request_timeout=ServicesCommonSettings().storage_service_upload_download_timeout,
+    )
 
-        if resp.error:
-            raise exceptions.S3TransferError(
-                "Error getting link: {}".format(resp.error.to_str())
-            )
-        if not resp.data.link:
-            raise exceptions.S3InvalidPathError(file_id)
-        log.debug("Got link %s", resp.data.link)
-        return URL(resp.data.link)
-    except ApiException as err:
-        _handle_api_exception(store_id, err)
+    if resp.error:
+        raise exceptions.S3TransferError(
+            "Error getting link: {}".format(resp.error.to_str())
+        )
+    if not resp.data.link:
+        raise exceptions.S3InvalidPathError(file_id)
+    log.debug("Got link %s", resp.data.link)
+    return URL(resp.data.link)
 
 
 async def _get_download_link(store_id: int, file_id: str, api: UsersApi) -> URL:
-    return await _get_link(store_id, file_id, api.download_file)
+    try:
+        return await _get_link(store_id, file_id, api.download_file)
+    except ApiException as err:
+        if err.status == 404:
+            raise exceptions.InvalidDownloadLinkError(None) from err
+        _handle_api_exception(store_id, err)
 
 
 async def _get_upload_link(store_id: int, file_id: str, api: UsersApi) -> URL:
-    return await _get_link(store_id, file_id, api.upload_file)
+    try:
+        return await _get_link(store_id, file_id, api.upload_file)
+    except ApiException as err:
+        _handle_api_exception(store_id, err)
 
 
 async def _download_link_to_file(session: ClientSession, url: URL, file_path: Path):

packages/simcore-sdk/tests/integration/conftest.py

@@ -11,6 +11,8 @@
 
 import np_helpers
 import pytest
+import requests
+import simcore_service_storage_sdk
 import sqlalchemy as sa
 from pytest_simcore.helpers.rawdata_fakers import random_project, random_user
 from simcore_postgres_database.storage_models import projects, users
@@ -88,12 +90,8 @@ async def filemanager_cfg(
 
 @pytest.fixture
 def create_valid_file_uuid(project_id: str, node_uuid: str) -> Callable:
-    def create(file_path: Path, project: str = None, node: str = None):
-        if project is None:
-            project = project_id
-        if node is None:
-            node = node_uuid
-        return np_helpers.file_uuid(file_path, project, node)
+    def create(file_path: Path):
+        return np_helpers.file_uuid(file_path, project_id, node_uuid)
 
     return create
 
@@ -131,18 +129,39 @@ def create_node_link(key: str) -> Dict[str, str]:
 
 @pytest.fixture()
 def store_link(
-    minio_service, bucket, create_valid_file_uuid, s3_simcore_location
+    bucket: str,  # packages/pytest-simcore/src/pytest_simcore/minio_service.py
+    create_valid_file_uuid: Callable,
+    s3_simcore_location: str,
+    user_id: int,
+    project_id: str,
+    node_uuid: str,
+    storage_service: URL,  # packages/pytest-simcore/src/pytest_simcore/simcore_storage_service.py
 ) -> Callable:
-    def create_store_link(
-        file_path: Path, project_id: str = None, node_id: str = None
-    ) -> Dict[str, str]:
-        # upload the file to S3
-        assert Path(file_path).exists()
-        file_id = create_valid_file_uuid(file_path, project_id, node_id)
-        # using the s3 client the path must be adapted
-        # TODO: use the storage sdk instead
-        s3_object = Path(project_id, node_id, Path(file_path).name).as_posix()
-        minio_service.upload_file(bucket, s3_object, str(file_path))
+    async def create_store_link(file_path: Path) -> Dict[str, str]:
+        file_path = Path(file_path)
+        assert file_path.exists()
+
+        file_id = create_valid_file_uuid(file_path)
+
+        # Get upload presigned link via storage-sdk API (both pg and s3 get updated)
+        user_api = simcore_service_storage_sdk.UsersApi()
+        user_api.api_client.configuration.host = str(storage_service.with_path("/v0"))
+
+        r: simcore_service_storage_sdk.PresignedLinkEnveloped = (
+            await user_api.upload_file(file_id, s3_simcore_location, user_id)
+        )
+
+        # Upload using the link
+        extra_hdr = {
+            "Content-Length": f"{file_path.stat().st_size}",
+            "Content-Type": "application/binary",
+        }
+        upload = requests.put(
+            r.data.link, data=file_path.read_bytes(), headers=extra_hdr
+        )
+        assert upload.status_code == 200, upload.text
+
+        # FIXME: that at this point, S3 and pg have some data that is NOT cleaned up
         return {"store": s3_simcore_location, "path": file_id}
 
     yield create_store_link
@@ -189,7 +208,7 @@ def special_2nodes_configuration(
     empty_configuration_file: Path,
     project_id: str,
     node_uuid: str,
-):
+) -> Callable:
     def create_config(
         prev_node_inputs: List[Tuple[str, str, Any]] = None,
         prev_node_outputs: List[Tuple[str, str, Any]] = None,
@@ -228,6 +247,7 @@ def create_config(
         return config_dict, project_id, node_uuid
 
     yield create_config
+
     # teardown
     postgres_session.query(ComputationalTask).delete()
     postgres_session.query(ComputationalPipeline).delete()