✨Storage can return a S3 link instead of a presigned link on demand (ITISFoundation#3004)

Author: sanderegg

api/specs/storage/openapi.yaml

@@ -292,6 +292,16 @@ paths:
           required: true
           schema:
             type: string
+        - name: link_type
+          in: query
+          required: false
+          schema:
+            type: string
+            default: "presigned"
+            enum:
+              - presigned
+              - s3
+
       responses:
         "200":
           description: "Returns presigned link"
@@ -330,6 +340,15 @@ paths:
           required: false
           schema:
             type: string
+        - name: link_type
+          in: query
+          required: false
+          schema:
+            type: string
+            default: "presigned"
+            enum:
+              - presigned
+              - s3
       responses:
         "200":
           description: "Returns presigned link"

services/storage/src/simcore_service_storage/api/v0/openapi.yaml

@@ -285,6 +285,15 @@ paths:
           required: true
           schema:
             type: string
+        - name: link_type
+          in: query
+          required: false
+          schema:
+            type: string
+            default: presigned
+            enum:
+              - presigned
+              - s3
       responses:
         '200':
           description: Returns presigned link
@@ -323,6 +332,15 @@ paths:
           required: false
           schema:
             type: string
+        - name: link_type
+          in: query
+          required: false
+          schema:
+            type: string
+            default: presigned
+            enum:
+              - presigned
+              - s3
       responses:
         '200':
           description: Returns presigned link

services/storage/src/simcore_service_storage/dsm.py

@@ -8,6 +8,7 @@
 import os
 import re
 import tempfile
+import urllib.parse
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
@@ -16,12 +17,14 @@
 
 import attr
 import botocore
+import botocore.exceptions
 import sqlalchemy as sa
 from aiobotocore.client import AioBaseClient
 from aiobotocore.session import AioSession, ClientCreatorContext, get_session
 from aiohttp import web
 from aiopg.sa import Engine
 from aiopg.sa.result import ResultProxy, RowProxy
+from pydantic import AnyUrl, parse_obj_as
 from servicelib.aiohttp.aiopg_utils import DBAPIError, PostgresRetryPolicyUponOperation
 from servicelib.aiohttp.client_session import get_client_session
 from servicelib.utils import fire_and_forget_task
@@ -493,7 +496,7 @@ async def auto_update_database_from_storage_task(
             file_uuid, bucket_name, object_name, silence_exception=True
         )
 
-    async def upload_link(self, user_id: str, file_uuid: str):
+    async def upload_link(self, user_id: str, file_uuid: str, as_presigned_link: bool):
         """
         Creates pre-signed upload link and updates metadata table when
         link is used and upload is successfuly completed
@@ -545,9 +548,16 @@ async def _init_metadata() -> Tuple[int, str]:
                 object_name=object_name,
             )
         )
-        return self.s3_client.create_presigned_put_url(bucket_name, object_name)
+        link = parse_obj_as(
+            AnyUrl, f"s3://{bucket_name}/{urllib.parse.quote( object_name)}"
+        )
+        if as_presigned_link:
+            link = self.s3_client.create_presigned_put_url(bucket_name, object_name)
+        return f"{link}"
 
-    async def download_link_s3(self, file_uuid: str, user_id: int) -> str:
+    async def download_link_s3(
+        self, file_uuid: str, user_id: int, as_presigned_link: bool
+    ) -> str:
 
         # access layer
         async with self.engine.acquire() as conn:
@@ -577,9 +587,12 @@ async def download_link_s3(self, file_uuid: str, user_id: int) -> str:
                 raise web.HTTPNotFound(
                     reason=f"File '{file_uuid}' does not exists in storage."
                 )
-
-        link = self.s3_client.create_presigned_get_url(bucket_name, object_name)
-        return link
+        link = parse_obj_as(
+            AnyUrl, f"s3://{bucket_name}/{urllib.parse.quote( object_name)}"
+        )
+        if as_presigned_link:
+            link = self.s3_client.create_presigned_get_url(bucket_name, object_name)
+        return f"{link}"
 
     async def download_link_datcore(self, user_id: str, file_id: str) -> URL:
         api_token, api_secret = self._get_datcore_tokens(user_id)
@@ -592,17 +605,23 @@ async def download_link_datcore(self, user_id: str, file_id: str) -> URL:
 
     # COPY -----------------------------
 
-    async def copy_file_s3_s3(self, user_id: str, dest_uuid: str, source_uuid: str):
+    async def copy_file_s3_s3(
+        self, user_id: str, dest_uuid: str, source_uuid: str
+    ) -> None:
         # FIXME: operation MUST be atomic
 
         # source is s3, location is s3
         to_bucket_name = self.simcore_bucket_name
         to_object_name = dest_uuid
         from_bucket = self.simcore_bucket_name
         from_object_name = source_uuid
-        # FIXME: This is not async!
-        self.s3_client.copy_object(
-            to_bucket_name, to_object_name, from_bucket, from_object_name
+        await asyncio.get_event_loop().run_in_executor(
+            None,
+            self.s3_client.copy_object,
+            to_bucket_name,
+            to_object_name,
+            from_bucket,
+            from_object_name,
         )
 
         # update db
@@ -616,6 +635,7 @@ async def copy_file_s3_s3(self, user_id: str, dest_uuid: str, source_uuid: str):
     async def copy_file_s3_datcore(
         self, user_id: str, dest_uuid: str, source_uuid: str
     ):
+        assert self.app  # nosec
         session = get_client_session(self.app)
 
         # source is s3, get link and copy to datcore
@@ -648,6 +668,7 @@ async def copy_file_datcore_s3(
         source_uuid: str,
         filename_missing: bool = False,
     ):
+        assert self.app  # nosec
         session = get_client_session(self.app)
 
         # 2 steps: Get download link for local copy, the upload link to s3
@@ -658,7 +679,9 @@ async def copy_file_datcore_s3(
         if filename_missing:
             dest_uuid = str(Path(dest_uuid) / filename)
 
-        s3_upload_link = await self.upload_link(user_id, dest_uuid)
+        s3_upload_link = await self.upload_link(
+            user_id, dest_uuid, as_presigned_link=True
+        )
 
         with tempfile.TemporaryDirectory() as tmpdir:
             # FIXME: connect download and upload streams
@@ -690,7 +713,7 @@ async def copy_file(
         dest_uuid: str,
         source_location: str,
         source_uuid: str,
-    ):
+    ) -> None:
         if source_location == SIMCORE_S3_STR:
             if dest_location == DATCORE_STR:
                 await self.copy_file_s3_datcore(user_id, dest_uuid, source_uuid)

services/storage/src/simcore_service_storage/handlers.py

@@ -299,6 +299,7 @@ async def download_file(request: web.Request):
     assert params["location_id"]  # nosec
     assert params["fileId"]  # nosec
     assert query["user_id"]  # nosec
+    link_type = query.get("link_type", "presigned")
 
     with handle_storage_errors():
         location_id = params["location_id"]
@@ -308,7 +309,9 @@ async def download_file(request: web.Request):
         dsm = await _prepare_storage_manager(params, query, request)
         location = dsm.location_from_id(location_id)
         if location == SIMCORE_S3_STR:
-            link = await dsm.download_link_s3(file_uuid, user_id)
+            link = await dsm.download_link_s3(
+                file_uuid, user_id, as_presigned_link=bool(link_type == "presigned")
+            )
         else:
             link = await dsm.download_link_datcore(user_id, file_uuid)
 
@@ -318,10 +321,11 @@ async def download_file(request: web.Request):
 @routes.put(f"/{api_vtag}/locations/{{location_id}}/files/{{fileId}}")  # type: ignore
 async def upload_file(request: web.Request):
     params, query, body = await extract_and_validate(request)
-
+    log.debug("received call to upload_file with %s", f"{params=}, {query=}, {body=}")
     assert params, "params %s" % params  # nosec
     assert query, "query %s" % query  # nosec
     assert not body, "body %s" % body  # nosec
+    link_type = query.get("link_type", "presigned")
 
     with handle_storage_errors():
         location_id = params["location_id"]
@@ -335,6 +339,7 @@ async def upload_file(request: web.Request):
             source_uuid = query["extra_source"]
             source_id = query["extra_location"]
             source_location = dsm.location_from_id(source_id)
+            # FIXME: this does not even return a link... nobody is using this??
             link = await dsm.copy_file(
                 user_id=user_id,
                 dest_location=location,
@@ -343,7 +348,11 @@ async def upload_file(request: web.Request):
                 source_uuid=source_uuid,
             )
         else:
-            link = await dsm.upload_link(user_id=user_id, file_uuid=file_uuid)
+            link = await dsm.upload_link(
+                user_id=user_id,
+                file_uuid=file_uuid,
+                as_presigned_link=bool(link_type == "presigned"),
+            )
 
     return {"error": None, "data": {"link": link}}
 

services/storage/tests/unit/test_dsm.py

@@ -143,7 +143,9 @@ def _create_file_meta_for_s3(
 async def _upload_file(
     dsm: DataStorageManager, file_metadata: FileMetaData, file_path: Path
 ) -> FileMetaData:
-    up_url = await dsm.upload_link(file_metadata.user_id, file_metadata.file_uuid)
+    up_url = await dsm.upload_link(
+        file_metadata.user_id, file_metadata.file_uuid, as_presigned_link=True
+    )
     assert file_path.exists()
     with file_path.open("rb") as fp:
         d = fp.read()
@@ -254,7 +256,9 @@ async def test_links_s3(
 
     tmp_file2 = f"{tmp_file}.rec"
     user_id = 0
-    down_url = await dsm.download_link_s3(fmd.file_uuid, user_id)
+    down_url = await dsm.download_link_s3(
+        fmd.file_uuid, user_id, as_presigned_link=True
+    )
 
     urllib.request.urlretrieve(down_url, tmp_file2)
 
@@ -276,7 +280,7 @@ async def test_copy_s3_s3(
     assert len(data) == 0
 
     # upload the file
-    up_url = await dsm.upload_link(fmd.user_id, fmd.file_uuid)
+    up_url = await dsm.upload_link(fmd.user_id, fmd.file_uuid, as_presigned_link=True)
     with tmp_file.open("rb") as fp:
         d = fp.read()
         req = urllib.request.Request(up_url, data=d, method="PUT")
@@ -350,7 +354,7 @@ async def test_dsm_s3_to_datcore(
 
     dsm = dsm_fixture
 
-    up_url = await dsm.upload_link(fmd.user_id, fmd.file_uuid)
+    up_url = await dsm.upload_link(fmd.user_id, fmd.file_uuid, as_presigned_link=True)
     with tmp_file.open("rb") as fp:
         d = fp.read()
         req = urllib.request.Request(up_url, data=d, method="PUT")
@@ -360,20 +364,22 @@ async def test_dsm_s3_to_datcore(
     # given the fmd, upload to datcore
     tmp_file2 = f"{tmp_file}.fordatcore"
     user_id = USER_ID
-    down_url = await dsm.download_link_s3(fmd.file_uuid)
+    down_url = await dsm.download_link_s3(
+        fmd.file_uuid, user_id, as_presigned_link=True
+    )
     urllib.request.urlretrieve(down_url, tmp_file2)
     assert filecmp.cmp(tmp_file2, tmp_file)
     # now we have the file locally, upload the file
     await dsm.upload_file_to_datcore(
-        user_id=user_id,
-        local_file_path=tmp_file2,
-        destination_id=datcore_structured_testbucket["dataset_id"],
+        user_id,
+        tmp_file2,
+        datcore_structured_testbucket["dataset_id"],
     )
     # and into a deeper strucutre
     await dsm.upload_file_to_datcore(
-        user_id=user_id,
-        local_file_path=tmp_file2,
-        destination_id=datcore_structured_testbucket["coll2_id"],
+        user_id,
+        tmp_file2,
+        datcore_structured_testbucket["coll2_id"],
     )
 
     # FIXME: upload takes some time
@@ -460,7 +466,7 @@ async def test_dsm_datcore_to_S3(
 
     # and the one on s3
     tmp_file2 = f"{tmp_file}.fromS3"
-    down_url_s3 = await dsm.download_link_s3(dest_uuid)
+    down_url_s3 = await dsm.download_link_s3(dest_uuid, user_id, as_presigned_link=True)
     urllib.request.urlretrieve(down_url_s3, tmp_file2)
 
     assert filecmp.cmp(tmp_file1, tmp_file2)
@@ -486,7 +492,7 @@ async def test_copy_datcore(
     tmp_file = mock_files_factory(1)[0]
     fmd = _create_file_meta_for_s3(postgres_service_url, s3_client, tmp_file)
 
-    up_url = await dsm.upload_link(fmd.user_id, fmd.file_uuid)
+    up_url = await dsm.upload_link(fmd.user_id, fmd.file_uuid, as_presigned_link=True)
     with tmp_file.open("rb") as fp:
         d = fp.read()
         req = urllib.request.Request(up_url, data=d, method="PUT")