✨ Add exporter code to storage (#7218)

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

api/specs/web-server/_long_running_tasks.py

@@ -13,7 +13,7 @@
 from servicelib.long_running_tasks._models import TaskGet, TaskStatus
 from simcore_service_webserver._meta import API_VTAG
 from simcore_service_webserver.tasks._exception_handlers import (
-    _TO_HTTP_ERROR_MAP as data_export_http_error_map,
+    _TO_HTTP_ERROR_MAP as export_data_http_error_map,
 )
 
 router = APIRouter(
@@ -23,9 +23,9 @@
     ],
 )
 
-_data_export_responses: dict[int | str, dict[str, Any]] = {
+_export_data_responses: dict[int | str, dict[str, Any]] = {
     i.status_code: {"model": EnvelopedError}
-    for i in data_export_http_error_map.values()
+    for i in export_data_http_error_map.values()
 }
 
 
@@ -34,7 +34,7 @@
     response_model=Envelope[list[TaskGet]],
     name="list_tasks",
     description="Lists all long running tasks",
-    responses=_data_export_responses,
+    responses=_export_data_responses,
 )
 def get_async_jobs(): ...
 
@@ -44,7 +44,7 @@ def get_async_jobs(): ...
     response_model=Envelope[TaskStatus],
     name="get_task_status",
     description="Retrieves the status of a task",
-    responses=_data_export_responses,
+    responses=_export_data_responses,
 )
 def get_async_job_status(
     _path_params: Annotated[_PathParam, Depends()],
@@ -55,7 +55,7 @@ def get_async_job_status(
     "/tasks/{task_id}",
     name="cancel_and_delete_task",
     description="Cancels and deletes a task",
-    responses=_data_export_responses,
+    responses=_export_data_responses,
     status_code=status.HTTP_204_NO_CONTENT,
 )
 def abort_async_job(
@@ -67,7 +67,7 @@ def abort_async_job(
     "/tasks/{task_id}/result",
     name="get_task_result",
     description="Retrieves the result of a task",
-    responses=_data_export_responses,
+    responses=_export_data_responses,
 )
 def get_async_job_result(
     _path_params: Annotated[_PathParam, Depends()],

api/specs/web-server/_storage.py

@@ -36,7 +36,7 @@
 from simcore_service_webserver._meta import API_VTAG
 from simcore_service_webserver.storage.schemas import DatasetMetaData, FileMetaData
 from simcore_service_webserver.tasks._exception_handlers import (
-    _TO_HTTP_ERROR_MAP as data_export_http_error_map,
+    _TO_HTTP_ERROR_MAP as export_data_http_error_map,
 )
 
 router = APIRouter(
@@ -221,9 +221,9 @@ async def is_completed_upload_file(
 
 
 # data export
-_data_export_responses: dict[int | str, dict[str, Any]] = {
+_export_data_responses: dict[int | str, dict[str, Any]] = {
     i.status_code: {"model": EnvelopedError}
-    for i in data_export_http_error_map.values()
+    for i in export_data_http_error_map.values()
 }
 
 
@@ -232,7 +232,7 @@ async def is_completed_upload_file(
     response_model=Envelope[TaskGet],
     name="export_data",
     description="Export data",
-    responses=_data_export_responses,
+    responses=_export_data_responses,
 )
-async def export_data(data_export: DataExportPost, location_id: LocationID):
+async def export_data(export_data: DataExportPost, location_id: LocationID):
     """Trigger data export. Returns async job id for getting status and results"""

packages/aws-library/src/aws_library/s3/_constants.py

@@ -9,6 +9,9 @@
 MULTIPART_COPY_THRESHOLD: Final[ByteSize] = TypeAdapter(ByteSize).validate_python(
     "100MiB"
 )
+STREAM_READER_CHUNK_SIZE: Final[ByteSize] = TypeAdapter(ByteSize).validate_python(
+    "10MiB"
+)
 
 PRESIGNED_LINK_MAX_SIZE: Final[ByteSize] = TypeAdapter(ByteSize).validate_python("5GiB")
 S3_MAX_FILE_SIZE: Final[ByteSize] = TypeAdapter(ByteSize).validate_python("5TiB")

packages/aws-library/tests/test_s3_client.py

@@ -27,8 +27,8 @@
 from aiohttp import ClientSession
 from aws_library.s3._client import _AWS_MAX_ITEMS_PER_PAGE, S3ObjectKey, SimcoreS3API
 from aws_library.s3._constants import (
-    MULTIPART_COPY_THRESHOLD,
     MULTIPART_UPLOADS_MIN_TOTAL_SIZE,
+    STREAM_READER_CHUNK_SIZE,
 )
 from aws_library.s3._errors import (
     S3BucketInvalidError,
@@ -1902,7 +1902,7 @@ async def test_workflow_compress_s3_objects_and_local_files_in_a_single_archive_
                 get_zip_bytes_iter(
                     archive_entries,
                     progress_bar=progress_bar,
-                    chunk_size=MULTIPART_COPY_THRESHOLD,
+                    chunk_size=STREAM_READER_CHUNK_SIZE,
                 )
             ),
         )

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/exceptions.py

@@ -27,5 +27,5 @@ class JobAbortedError(BaseAsyncjobRpcError):
 
 class JobError(BaseAsyncjobRpcError):
     msg_template: str = (
-        "Job {job_id} failed with exception type {exc_type} and message {exc_msg}"
+        "Job '{job_id}' failed with exception type '{exc_type}' and message: {exc_msg}"
     )

packages/models-library/src/models_library/api_schemas_storage/data_export_async_jobs.py renamed to packages/models-library/src/models_library/api_schemas_storage/export_data_async_jobs.py

@@ -1,14 +1,6 @@
 # pylint: disable=R6301
 
 from common_library.errors_classes import OsparcErrorMixin
-from models_library.projects_nodes_io import LocationID, StorageFileID
-from pydantic import BaseModel, Field
-
-
-class DataExportTaskStartInput(BaseModel):
-    location_id: LocationID
-    file_and_folder_ids: list[StorageFileID] = Field(..., min_length=1)
-
 
 ### Exceptions
 

packages/models-library/src/models_library/api_schemas_webserver/storage.py

@@ -3,15 +3,12 @@
 
 from pydantic import BaseModel, Field
 
-from ..api_schemas_storage.data_export_async_jobs import DataExportTaskStartInput
 from ..api_schemas_storage.storage_schemas import (
     DEFAULT_NUMBER_OF_PATHS_PER_PAGE,
     MAX_NUMBER_OF_PATHS_PER_PAGE,
 )
-from ..projects_nodes_io import LocationID, StorageFileID
-from ..rest_pagination import (
-    CursorQueryParameters,
-)
+from ..projects_nodes_io import LocationID
+from ..rest_pagination import CursorQueryParameters
 from ._base import InputSchema
 
 
@@ -40,11 +37,8 @@ class BatchDeletePathsBodyParams(InputSchema):
     paths: set[Path]
 
 
-class DataExportPost(InputSchema):
-    paths: list[StorageFileID]
+PathToExport = Path
 
-    def to_rpc_schema(self, location_id: LocationID) -> DataExportTaskStartInput:
-        return DataExportTaskStartInput(
-            file_and_folder_ids=self.paths,
-            location_id=location_id,
-        )
+
+class DataExportPost(InputSchema):
+    paths: list[PathToExport]

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

@@ -4,6 +4,7 @@
 
 SEE tests_basic_regex.py for examples
 """
+
 # TODO: for every pattern we should have a formatter function
 # NOTE: some sites to manualy check ideas
 #   https://regex101.com/
@@ -45,7 +46,9 @@
 )
 
 # Storage basic file ID
-SIMCORE_S3_FILE_ID_RE = rf"^(api|({UUID_RE_BASE}))\/({UUID_RE_BASE})\/(.+)$"
+SIMCORE_S3_FILE_ID_RE = rf"^(exports\/\d+\/{UUID_RE_BASE}\.zip)|((api|({UUID_RE_BASE}))\/({UUID_RE_BASE})\/(.+)$)"
+
+
 SIMCORE_S3_DIRECTORY_ID_RE = rf"^({UUID_RE_BASE})\/({UUID_RE_BASE})\/(.+)\/$"
 
 # S3 - AWS bucket names [https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html]

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

@@ -95,9 +95,9 @@ class ProgressType(StrAutoEnum):
 
 
 class ProgressMessageMixin(RabbitMessageBase):
-    channel_name: Literal[
+    channel_name: Literal["simcore.services.progress.v2"] = (
         "simcore.services.progress.v2"
-    ] = "simcore.services.progress.v2"
+    )
     progress_type: ProgressType = (
         ProgressType.COMPUTATION_RUNNING
     )  # NOTE: backwards compatible
@@ -118,9 +118,9 @@ def routing_key(self) -> str | None:
 
 
 class InstrumentationRabbitMessage(RabbitMessageBase, NodeMessageBase):
-    channel_name: Literal[
+    channel_name: Literal["simcore.services.instrumentation"] = (
         "simcore.services.instrumentation"
-    ] = "simcore.services.instrumentation"
+    )
     metrics: str
     service_uuid: NodeID
     service_type: str
@@ -210,9 +210,9 @@ def routing_key(self) -> str | None:
 
 
 class RabbitResourceTrackingStartedMessage(RabbitResourceTrackingBaseMessage):
-    message_type: Literal[
+    message_type: Literal[RabbitResourceTrackingMessageType.TRACKING_STARTED] = (
         RabbitResourceTrackingMessageType.TRACKING_STARTED
-    ] = RabbitResourceTrackingMessageType.TRACKING_STARTED
+    )
 
     wallet_id: WalletID | None
     wallet_name: str | None
@@ -250,9 +250,9 @@ class RabbitResourceTrackingStartedMessage(RabbitResourceTrackingBaseMessage):
 
 
 class RabbitResourceTrackingHeartbeatMessage(RabbitResourceTrackingBaseMessage):
-    message_type: Literal[
+    message_type: Literal[RabbitResourceTrackingMessageType.TRACKING_HEARTBEAT] = (
         RabbitResourceTrackingMessageType.TRACKING_HEARTBEAT
-    ] = RabbitResourceTrackingMessageType.TRACKING_HEARTBEAT
+    )
 
 
 class SimcorePlatformStatus(StrAutoEnum):
@@ -261,9 +261,9 @@ class SimcorePlatformStatus(StrAutoEnum):
 
 
 class RabbitResourceTrackingStoppedMessage(RabbitResourceTrackingBaseMessage):
-    message_type: Literal[
+    message_type: Literal[RabbitResourceTrackingMessageType.TRACKING_STOPPED] = (
         RabbitResourceTrackingMessageType.TRACKING_STOPPED
-    ] = RabbitResourceTrackingMessageType.TRACKING_STOPPED
+    )
 
     simcore_platform_status: SimcorePlatformStatus = Field(
         ...,
@@ -297,9 +297,9 @@ class CreditsLimit(IntEnum):
 
 
 class WalletCreditsLimitReachedMessage(RabbitMessageBase):
-    channel_name: Literal[
+    channel_name: Literal["io.simcore.service.wallets-credit-limit-reached"] = (
         "io.simcore.service.wallets-credit-limit-reached"
-    ] = "io.simcore.service.wallets-credit-limit-reached"
+    )
     created_at: datetime.datetime = Field(
         default_factory=lambda: arrow.utcnow().datetime,
         description="message creation datetime",

packages/models-library/tests/test_project_nodes_io.py

@@ -2,7 +2,8 @@
 # pylint:disable=unused-argument
 # pylint:disable=redefined-outer-name
 
-from typing import Any
+from typing import Any, Final
+from uuid import UUID
 
 import pytest
 from faker import Faker
@@ -11,9 +12,14 @@
     DatCoreFileLink,
     SimCoreFileLink,
     SimcoreS3DirectoryID,
+    SimcoreS3FileID,
 )
+from models_library.users import UserID
 from pydantic import TypeAdapter, ValidationError
 
+UUID_0: Final[str] = f"{UUID(int=0)}"
+USER_ID_0: Final[UserID] = 0
+
 
 @pytest.fixture()
 def minimal_simcore_file_link(faker: Faker) -> dict[str, Any]:
@@ -115,9 +121,6 @@ def test_store_discriminator():
     assert isinstance(rawgraph_node.inputs["input_1"], PortLink)
 
 
-UUID_0: str = "00000000-0000-0000-0000-000000000000"
-
-
 def test_simcore_s3_directory_id():
     # the only allowed path is the following
     result = TypeAdapter(SimcoreS3DirectoryID).validate_python(
@@ -180,3 +183,18 @@ def test_simcore_s3_directory_get_parent():
         SimcoreS3DirectoryID._get_parent(  # noqa SLF001
             "/hello/object/", parent_index=4
         )
+
+
+@pytest.mark.parametrize(
+    "object_key",
+    [
+        f"api/{UUID_0}/some-random-file.png",
+        f"exports/{USER_ID_0}/{UUID_0}.zip",
+        f"{UUID_0}/{UUID_0}/some-random-file.png",
+        f"api/{UUID_0}/some-path/some-random-file.png",
+        f"{UUID_0}/{UUID_0}/some-path/some-random-file.png",
+    ],
+)
+def test_simcore_s3_file_id_accepted_patterns(object_key: str):
+    file_id = TypeAdapter(SimcoreS3FileID).validate_python(object_key)
+    assert f"{file_id}" == object_key

packages/models-library/tests/test_rabbit_messages.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import pytest
 from faker import Faker
 from models_library.progress_bar import ProgressReport
@@ -41,6 +39,6 @@
 )
 async def test_raw_message_parsing(raw_data: str, class_type: type):
     result = TypeAdapter(
-        Union[ProgressRabbitMessageNode, ProgressRabbitMessageProject]
+        ProgressRabbitMessageNode | ProgressRabbitMessageProject
     ).validate_json(raw_data)
-    assert type(result) == class_type
+    assert type(result) is class_type

packages/service-library/src/servicelib/archiving_utils/_tdqm_utils.py

@@ -1,8 +1,7 @@
 from typing import Final
 
-from pydantic import ByteSize, NonNegativeFloat, NonNegativeInt
+from pydantic import ByteSize, NonNegativeInt
 
-_UNIT_MULTIPLIER: Final[NonNegativeFloat] = 1024.0
 TQDM_FILE_OPTIONS: Final[dict] = {
     "unit": "byte",
     "unit_scale": True,

packages/service-library/src/servicelib/bytes_iters/_stream_zip.py

@@ -1,14 +1,17 @@
+import logging
 from collections.abc import AsyncIterable
 from datetime import UTC, datetime
 from stat import S_IFREG
 from typing import TypeAlias
 
 from models_library.bytes_iters import BytesIter, DataSize
-from stream_zip import ZIP_32, AsyncMemberFile, async_stream_zip
+from stream_zip import ZIP_64, AsyncMemberFile, async_stream_zip
 
 from ..progress_bar import ProgressBarData
 from ._models import BytesStreamer
 
+_logger = logging.getLogger(__name__)
+
 FileNameInArchive: TypeAlias = str
 ArchiveFileEntry: TypeAlias = tuple[FileNameInArchive, BytesStreamer]
 ArchiveEntries: TypeAlias = list[ArchiveFileEntry]
@@ -22,7 +25,7 @@ async def _member_files_iter(
             file_name,
             datetime.now(UTC),
             S_IFREG | 0o600,
-            ZIP_32,
+            ZIP_64,
             byte_streamer.with_progress_bytes_iter(progress_bar=progress_bar),
         )