♻️✨adding nodeports support to dynamic-sidecar

api/specs/webserver/openapi-projects.yaml

@@ -395,6 +395,55 @@ paths:
         default:
           $ref: "./openapi.yaml#/components/responses/DefaultErrorResponse"
 
+  /projects/{project_id}/nodes/{node_id}/retrieve:
+    parameters:
+      - name: project_id
+        in: path
+        required: true
+        schema:
+          type: string
+      - name: node_id
+        in: path
+        required: true
+        schema:
+          type: string
+
+    post:
+      tags:
+        - project
+      description: Triggers service retrieve
+      operationId: post_retrieve
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                port_keys:
+                  description: list of por keys to be retrieved
+                  type: array
+                  items:
+                    type: string
+      responses:
+        "200":
+          description: Returns the amount of transferred bytes when pulling data via nodeports
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  data:
+                    type: object
+                    description: response payload
+                    properties:
+                      size_bytes:
+                        type: integer
+                        description: amount of transferred bytes
+
+        default:
+          $ref: "#/components/responses/DefaultErrorResponse"
+
   /projects/{study_uuid}/tags/{tag_id}:
     parameters:
       - name: tag_id
@@ -716,7 +765,8 @@ components:
         created_at:
           title: Created At
           type: string
-          description: Timestamp of the time snapshot was taken from parent.Notice that
+          description:
+            Timestamp of the time snapshot was taken from parent.Notice that
             parent might change with time
           format: date-time
         parent_uuid:

api/specs/webserver/openapi.yaml

@@ -200,6 +200,9 @@ paths:
   /projects/{project_id}/nodes/{node_id}:
     $ref: "./openapi-projects.yaml#/paths/~1projects~1{project_id}~1nodes~1{node_id}"
 
+  /projects/{project_id}/nodes/{node_id}:retrieve:
+    $ref: "./openapi-projects.yaml#/paths/~1projects~1{project_id}~1nodes~1{node_id}~1retrieve"
+
   /nodes/{nodeInstanceUUID}/outputUi/{outputKey}:
     $ref: "./openapi-node-v0.0.1.yaml#/paths/~1nodes~1{nodeInstanceUUID}~1outputUi~1{outputKey}"
 

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

@@ -230,52 +230,54 @@ def jupyter_service(docker_registry: str, node_meta_schema: Dict) -> Dict[str, s
     )
 
 
-DY_STATIC_FILE_SERVER_VERSION = "1.0.5"
+@pytest.fixture(scope="session", params=["2.0.0"])
+def dy_static_file_server_version(request):
+    return request.param
 
 
 @pytest.fixture(scope="session")
 def dy_static_file_server_service(
-    docker_registry: str, node_meta_schema: Dict
+    docker_registry: str, node_meta_schema: Dict, dy_static_file_server_version: str
 ) -> Dict[str, str]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server
     """
     return _pull_push_service(
         "itisfoundation/dy-static-file-server",
-        DY_STATIC_FILE_SERVER_VERSION,
+        dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )
 
 
 @pytest.fixture(scope="session")
 def dy_static_file_server_dynamic_sidecar_service(
-    docker_registry: str, node_meta_schema: Dict
+    docker_registry: str, node_meta_schema: Dict, dy_static_file_server_version: str
 ) -> Dict[str, str]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server-dynamic-sidecar
     """
     return _pull_push_service(
         "itisfoundation/dy-static-file-server-dynamic-sidecar",
-        DY_STATIC_FILE_SERVER_VERSION,
+        dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )
 
 
 @pytest.fixture(scope="session")
 def dy_static_file_server_dynamic_sidecar_compose_spec_service(
-    docker_registry: str, node_meta_schema: Dict
+    docker_registry: str, node_meta_schema: Dict, dy_static_file_server_version: str
 ) -> Dict[str, str]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server-dynamic-sidecar-compose-spec
     """
     return _pull_push_service(
         "itisfoundation/dy-static-file-server-dynamic-sidecar-compose-spec",
-        DY_STATIC_FILE_SERVER_VERSION,
+        dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )

packages/service-library/src/servicelib/async_utils.py

@@ -1,22 +1,44 @@
 import asyncio
 from collections import deque
 from functools import wraps
-from typing import Dict, List
-
-import attr
-
-
-@attr.s(auto_attribs=True)
-class Context:
-    in_queue: asyncio.Queue
-    out_queue: asyncio.Queue
+from typing import TYPE_CHECKING, Any, Callable, Deque, Dict, List, Optional
+
+from pydantic import BaseModel
+
+if TYPE_CHECKING:
+    Queue = asyncio.Queue
+else:
+    # This is a bit of hack, but makes mypy happy.
+    # The issue is with asyncio.Queue which does not inherit from
+    # `typing.Generic`, but typeshed stubs for it says that it does.
+    # As a result there is no `__getitem__` inherited from
+    # typing.GenericMeta at runtime.
+    # SEE https://stackoverflow.com/a/48554601/2855718
+    class FakeGenericMeta(type):
+        def __getitem__(self, item):
+            return self
+
+    class Queue(
+        asyncio.Queue, metaclass=FakeGenericMeta
+    ):  # pylint: disable=function-redefined
+        pass
+
+
+class Context(BaseModel):
+    in_queue: Queue
+    out_queue: Queue
     initialized: bool
 
+    class Config:
+        arbitrary_types_allowed: bool = True
+
 
-sequential_jobs_contexts = {}
+_sequential_jobs_contexts: Dict[str, Context] = {}
 
 
-def run_sequentially_in_context(target_args: List[str] = None):
+def run_sequentially_in_context(
+    target_args: List[str] = None,
+) -> Callable[[Any], Any]:
     """All request to function with same calling context will be run sequentially.
 
     Example:
@@ -35,11 +57,13 @@ async def func(param1, param2, param3):
 
         functions = [
             func(1, "something", 3),
-            func(1, "else", 3),
+            func(1, "argument.attribute", 3),
             func(1, "here", 3),
         ]
         await asyncio.gather(*functions)
 
+    note the special "argument.attribute", which will use the attribute of argument to create the context.
+
     The following calls will run in parallel, because they have different contexts:
 
         functions = [
@@ -52,43 +76,55 @@ async def func(param1, param2, param3):
     """
     target_args = [] if target_args is None else target_args
 
-    def internal(decorated_function):
-        def get_context(args, kwargs: Dict) -> Context:
+    def internal(
+        decorated_function: Callable[[Any], Optional[Any]]
+    ) -> Callable[[Any], Optional[Any]]:
+        def get_context(args: Any, kwargs: Dict[Any, Any]) -> Context:
             arg_names = decorated_function.__code__.co_varnames[
                 : decorated_function.__code__.co_argcount
             ]
             search_args = dict(zip(arg_names, args))
             search_args.update(kwargs)
 
-            key_parts = deque()
+            key_parts: Deque[str] = deque()
             for arg in target_args:
-                if arg not in search_args:
+                sub_args = arg.split(".")
+                main_arg = sub_args[0]
+                if main_arg not in search_args:
                     message = (
-                        f"Expected '{arg}' in '{decorated_function.__name__}'"
+                        f"Expected '{main_arg}' in '{decorated_function.__name__}'"
                         f" arguments. Got '{search_args}'"
                     )
                     raise ValueError(message)
-                key_parts.append(search_args[arg])
+                context_key = search_args[main_arg]
+                for attribute in sub_args[1:]:
+                    potential_key = getattr(context_key, attribute)
+                    if not potential_key:
+                        message = f"Expected '{attribute}' attribute in '{context_key.__name__}' arguments."
+                        raise ValueError(message)
+                    context_key = potential_key
+
+                key_parts.append(f"{decorated_function.__name__}_{context_key}")
 
             key = ":".join(map(str, key_parts))
 
-            if key not in sequential_jobs_contexts:
-                sequential_jobs_contexts[key] = Context(
-                    in_queue=asyncio.Queue(),
-                    out_queue=asyncio.Queue(),
+            if key not in _sequential_jobs_contexts:
+                _sequential_jobs_contexts[key] = Context(
+                    in_queue=Queue(),
+                    out_queue=Queue(),
                     initialized=False,
                 )
 
-            return sequential_jobs_contexts[key]
+            return _sequential_jobs_contexts[key]
 
         @wraps(decorated_function)
-        async def wrapper(*args, **kwargs):
+        async def wrapper(*args: Any, **kwargs: Any) -> Any:
             context: Context = get_context(args, kwargs)
 
             if not context.initialized:
                 context.initialized = True
 
-                async def worker(in_q: asyncio.Queue, out_q: asyncio.Queue):
+                async def worker(in_q: Queue, out_q: Queue) -> None:
                     while True:
                         awaitable = await in_q.get()
                         in_q.task_done()
@@ -98,11 +134,9 @@ async def worker(in_q: asyncio.Queue, out_q: asyncio.Queue):
                             result = e
                         await out_q.put(result)
 
-                asyncio.get_event_loop().create_task(
-                    worker(context.in_queue, context.out_queue)
-                )
+                asyncio.create_task(worker(context.in_queue, context.out_queue))
 
-            await context.in_queue.put(decorated_function(*args, **kwargs))
+            await context.in_queue.put(decorated_function(*args, **kwargs))  # type: ignore
 
             wrapped_result = await context.out_queue.get()
             if isinstance(wrapped_result, Exception):

packages/service-library/src/servicelib/pools.py

@@ -1,5 +1,7 @@
+import asyncio
 from concurrent.futures import ProcessPoolExecutor
 from contextlib import contextmanager
+from typing import Any, Callable
 
 # only gets created on use and is guaranteed to be the s
 # ame for the entire lifetime of the application
@@ -35,3 +37,10 @@ def non_blocking_process_pool_executor(**kwargs) -> ProcessPoolExecutor:
         # FIXME: uncomment below line when the issue is fixed
         # executor.shutdown(wait=False)
         pass
+
+
+async def async_on_threadpool(callable_function: Callable, *args: Any) -> Any:
+    """Ensures blocking operation runs on shared thread pool"""
+    return await asyncio.get_event_loop().run_in_executor(
+        None, callable_function, *args
+    )

packages/service-library/src/servicelib/utils.py

@@ -7,7 +7,6 @@
 import asyncio
 import logging
 import os
-
 from pathlib import Path
 from typing import Any, Awaitable, Coroutine, List, Optional, Union
 
@@ -79,8 +78,11 @@ def log_exception_callback(fut: asyncio.Future):
 
 # // tasks
 async def logged_gather(
-    *tasks, reraise: bool = True, log: logging.Logger = logger, max_concurrency: int = 0
-) -> List[Any]:
+    *tasks: Awaitable[Any],
+    reraise: bool = True,
+    log: logging.Logger = logger,
+    max_concurrency: int = 0,
+) -> List[Optional[Any]]:
     """
         Thin wrapper around asyncio.gather that allows excuting ALL tasks concurently until the end
         even if any of them fail. Finally, all errors are logged and the first raised (if reraise=True)
@@ -91,18 +93,15 @@ async def logged_gather(
         use directly asyncio.gather(*tasks, return_exceptions=True).
 
     :param reraise: reraises first exception (in order the tasks were passed) concurrent tasks, defaults to True
-    :type reraise: bool, optional
     :param log: passing the logger gives a chance to identify the origin of the gather call, defaults to current submodule's logger
-    :type log: logging.Logger, optional
     :return: list of tasks results and errors e.g. [1, 2, ValueError("task3 went wrong"), 33, "foo"]
-    :rtype: List[Any]
     """
 
     wrapped_tasks = tasks
     if max_concurrency > 0:
         semaphore = asyncio.Semaphore(max_concurrency)
 
-        async def sem_task(task):
+        async def sem_task(task: Awaitable[Any]) -> Any:
             async with semaphore:
                 return await task
 

packages/service-library/tests/test_async_utils.py

@@ -8,14 +8,17 @@
 from typing import Any, Dict, List
 
 import pytest
-from servicelib.async_utils import run_sequentially_in_context, sequential_jobs_contexts
+from servicelib.async_utils import (
+    _sequential_jobs_contexts,
+    run_sequentially_in_context,
+)
 
 
 @pytest.fixture(autouse=True)
 def ensure_run_in_sequence_context_is_empty():
     # NOTE: since the contexts variable is initialized at import time, when several test run
     # the import happens only once and is rendered invalid, therefore explicit clearance is necessary
-    sequential_jobs_contexts.clear()
+    _sequential_jobs_contexts.clear()
 
 
 class LockedStore:

packages/service-library/tests/test_pools.py

@@ -1,8 +1,7 @@
 from asyncio import BaseEventLoop
 from concurrent.futures import ProcessPoolExecutor
 
-
-from servicelib.pools import non_blocking_process_pool_executor
+from servicelib.pools import async_on_threadpool, non_blocking_process_pool_executor
 
 
 def return_int_one() -> int:
@@ -32,3 +31,7 @@ async def test_different_pool_instances() -> None:
         max_workers=1
     ) as first, non_blocking_process_pool_executor() as second:
         assert first != second
+
+
+async def test_run_on_thread_pool() -> None:
+    assert await async_on_threadpool(return_int_one) == 1

packages/simcore-sdk/requirements/_test.in

@@ -18,6 +18,7 @@ pytest-mock
 pytest-runner
 pytest-sugar
 pytest-xdist
+pytest-lazy-fixture
 
 # mockups/fixtures
 alembic