✨Computation of path size: WebAPI + RPC to storage through Asyncjobs (ITISFoundation#7315)

Author: sanderegg

api/specs/web-server/_storage.py

@@ -26,6 +26,7 @@
     StorageAsyncJobResult,
     StorageAsyncJobStatus,
     StorageLocationPathParams,
+    StoragePathComputeSizeParams,
 )
 from models_library.generics import Envelope
 from models_library.projects_nodes_io import LocationID
@@ -68,6 +69,15 @@ async def list_storage_paths(
     """Lists the files/directories in WorkingDirectory"""
 
 
+@router.post(
+    "/storage/locations/{location_id}/paths/{path}:size",
+    response_model=Envelope[StorageAsyncJobGet],
+    status_code=status.HTTP_202_ACCEPTED,
+)
+async def compute_path_size(_path: Annotated[StoragePathComputeSizeParams, Depends()]):
+    """Compute the size of a path"""
+
+
 @router.get(
     "/storage/locations/{location_id}/datasets",
     response_model=Envelope[list[DatasetMetaData]],

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

@@ -1,10 +1,6 @@
 from pathlib import Path
 from typing import Annotated, Any
 
-from models_library.api_schemas_storage.storage_schemas import (
-    DEFAULT_NUMBER_OF_PATHS_PER_PAGE,
-    MAX_NUMBER_OF_PATHS_PER_PAGE,
-)
 from pydantic import BaseModel, Field
 
 from ..api_schemas_rpc_async_jobs.async_jobs import (
@@ -14,6 +10,10 @@
     AsyncJobStatus,
 )
 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 ..progress_bar import ProgressReport
 from ..projects_nodes_io import LocationID, StorageFileID
 from ..rest_pagination import (
@@ -26,6 +26,10 @@ class StorageLocationPathParams(BaseModel):
     location_id: LocationID
 
 
+class StoragePathComputeSizeParams(StorageLocationPathParams):
+    path: Path
+
+
 class ListPathsQueryParams(InputSchema, CursorQueryParameters):
     file_filter: Path | None = None
 

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/async_jobs/async_jobs.py

@@ -23,7 +23,7 @@ async def abort(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> AsyncJobAbort:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -41,7 +41,7 @@ async def get_status(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> AsyncJobStatus:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -59,7 +59,7 @@ async def get_result(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> AsyncJobResult:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -77,7 +77,7 @@ async def list_jobs(
     *,
     rpc_namespace: RPCNamespace,
     filter_: str,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> list[AsyncJobGet]:
     result: list[AsyncJobGet] = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -95,7 +95,7 @@ async def submit_job(
     rpc_namespace: RPCNamespace,
     method_name: str,
     job_id_data: AsyncJobNameData,
-    **kwargs
+    **kwargs,
 ) -> AsyncJobGet:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -104,5 +104,5 @@ async def submit_job(
         **kwargs,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
-    assert isinstance(result, AsyncJobGet)
+    assert isinstance(result, AsyncJobGet)  # nosec
     return result

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/storage/__init__.py

@@ -0,0 +1,33 @@
+from pathlib import Path
+
+from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobNameData,
+)
+from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
+from models_library.api_schemas_webserver.storage import StorageAsyncJobGet
+from models_library.projects_nodes_io import LocationID
+from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.users import UserID
+
+from ..._client_rpc import RabbitMQRPCClient
+from ..async_jobs.async_jobs import submit_job
+
+
+async def compute_path_size(
+    client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    product_name: str,
+    location_id: LocationID,
+    path: Path,
+) -> tuple[StorageAsyncJobGet, AsyncJobNameData]:
+    job_id_data = AsyncJobNameData(user_id=user_id, product_name=product_name)
+    async_job_rpc_get = await submit_job(
+        rabbitmq_rpc_client=client,
+        rpc_namespace=STORAGE_RPC_NAMESPACE,
+        method_name=RPCMethodName("compute_path_size"),
+        job_id_data=job_id_data,
+        location_id=location_id,
+        path=path,
+    )
+    return StorageAsyncJobGet.from_rpc_schema(async_job_rpc_get), job_id_data

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/storage/paths.py

@@ -42,10 +42,32 @@ APP_LOG_LEVEL=${STORAGE_LOGLEVEL:-${LOG_LEVEL:-${LOGLEVEL:-INFO}}}
 SERVER_LOG_LEVEL=$(echo "${APP_LOG_LEVEL}" | tr '[:upper:]' '[:lower:]')
 echo "$INFO" "Log-level app/server: $APP_LOG_LEVEL/$SERVER_LOG_LEVEL"
 
-if [ "${SC_BOOT_MODE}" = "debug" ]; then
-  reload_dir_packages=$(find /devel/packages -maxdepth 3 -type d -path "*/src/*" ! -path "*.*" -exec echo '--reload-dir {} \' \;)
+if [ "${STORAGE_WORKER_MODE}" = "true" ]; then
+  if [ "${SC_BOOT_MODE}" = "debug" ]; then
+    exec watchmedo auto-restart \
+      --directory /devel/packages \
+      --directory services/storage \
+      --pattern "*.py" \
+      --recursive \
+      -- \
+      celery \
+      --app=simcore_service_storage.modules.celery.worker_main:app \
+      worker --pool=threads \
+      --loglevel="${SERVER_LOG_LEVEL}" \
+      --concurrency="${CELERY_CONCURRENCY}"
+  else
+    exec celery \
+      --app=simcore_service_storage.modules.celery.worker_main:app \
+      worker --pool=threads \
+      --loglevel="${SERVER_LOG_LEVEL}" \
+      --hostname="${HOSTNAME}" \
+      --concurrency="${CELERY_CONCURRENCY}"
+  fi
+else
+  if [ "${SC_BOOT_MODE}" = "debug" ]; then
+    reload_dir_packages=$(find /devel/packages -maxdepth 3 -type d -path "*/src/*" ! -path "*.*" -exec echo '--reload-dir {} \' \;)
 
-  exec sh -c "
+    exec sh -c "
     cd services/storage/src/simcore_service_storage && \
     python -m debugpy --listen 0.0.0.0:${STORAGE_REMOTE_DEBUGGING_PORT} -m uvicorn main:app \
       --host 0.0.0.0 \
@@ -55,14 +77,6 @@ if [ "${SC_BOOT_MODE}" = "debug" ]; then
       --reload-dir . \
       --log-level \"${SERVER_LOG_LEVEL}\"
   "
-else
-  if [ "${STORAGE_WORKER_MODE}" = "true" ]; then
-    exec celery \
-      --app=simcore_service_storage.modules.celery.worker_main:app \
-      worker --pool=threads \
-      --loglevel="${SERVER_LOG_LEVEL}" \
-      --hostname="${HOSTNAME}" \
-      --concurrency="${CELERY_CONCURRENCY}"
   else
     exec uvicorn simcore_service_storage.main:app \
       --host 0.0.0.0 \

services/storage/docker/boot.sh

@@ -29,3 +29,4 @@ pydantic[dotenv]
 tenacity
 typer
 types-aiobotocore[s3] # s3 storage
+watchdog

services/storage/requirements/_base.in

@@ -984,6 +984,8 @@ vine==5.1.0
     #   amqp
     #   celery
     #   kombu
+watchdog==6.0.0
+    # via -r requirements/_base.in
 watchfiles==1.0.4
     # via uvicorn
 wcwidth==0.2.13

services/storage/requirements/_base.txt

@@ -374,7 +374,6 @@ setuptools==75.8.2
     # via
     #   moto
     #   pytest-celery
-    # via moto
 simcore-service-storage-sdk @ git+https://github.com/ITISFoundation/osparc-simcore.git@cfdf4f86d844ebb362f4f39e9c6571d561b72897#subdirectory=services/storage/client-sdk/python
     # via -r requirements/_test.in
 six==1.17.0

services/storage/requirements/_test.txt

@@ -87,6 +87,8 @@ typing-extensions==4.12.2
 virtualenv==20.29.2
     # via pre-commit
 watchdog==6.0.0
-    # via -r requirements/_tools.in
+    # via
+    #   -c requirements/_base.txt
+    #   -r requirements/_tools.in
 wheel==0.45.1
     # via pip-tools