moved naming

Author: sanderegg

packages/dask-task-models-library/src/dask_task_models_library/resource_constraints.py

@@ -1,9 +1,11 @@
-from typing import Literal, TypedDict
+from typing import Final, Literal, TypedDict
 
 from .constants import DASK_TASK_EC2_RESOURCE_RESTRICTION_KEY
 
+DASK_WORKER_THREAD_RESOURCE_NAME: Final[str] = "threads"
 
-class DaskTaskResources(TypedDict):
+
+class DaskTaskResources(TypedDict, total=False):
     CPU: float
     RAM: int  # in bytes
     # threads is a constant of 1 (enforced by static type checkers via Literal)

services/autoscaling/src/simcore_service_autoscaling/modules/dask.py

@@ -9,7 +9,10 @@
 import dask.typing
 import distributed
 from aws_library.ec2 import EC2InstanceData, Resources
-from dask_task_models_library.resource_constraints import DaskTaskResources
+from dask_task_models_library.resource_constraints import (
+    DASK_WORKER_THREAD_RESOURCE_NAME,
+    DaskTaskResources,
+)
 from distributed.core import Status
 from models_library.clusters import ClusterAuthentication, TLSAuthentication
 from pydantic import AnyUrl, ByteSize, TypeAdapter
@@ -39,7 +42,6 @@ async def _wrap_client_async_routine(
 
 
 _DASK_SCHEDULER_CONNECT_TIMEOUT_S: Final[int] = 5
-DASK_WORKER_THREAD_RESOURCE_NAME: Final[str] = "threads"
 
 
 @contextlib.asynccontextmanager

services/autoscaling/tests/unit/test_modules_dask.py

@@ -11,6 +11,9 @@
 import pytest
 from arrow import utcnow
 from aws_library.ec2 import Resources
+from dask_task_models_library.resource_constraints import (
+    DASK_WORKER_THREAD_RESOURCE_NAME,
+)
 from faker import Faker
 from models_library.clusters import (
     ClusterAuthentication,
@@ -31,7 +34,6 @@
     EC2InstanceData,
 )
 from simcore_service_autoscaling.modules.dask import (
-    DASK_WORKER_THREAD_RESOURCE_NAME,
     DaskMonitoringSettings,
     DaskTask,
     _scheduler_client,
@@ -122,15 +124,13 @@ async def test_list_unrunnable_tasks(
     # we have nothing running now
     assert await list_unrunnable_tasks(scheduler_url, scheduler_authentication) == []
     # start a task that cannot run
-    dask_task_impossible_resources = {"XRAM": 213}
+    dask_task_impossible_resources = DaskTaskResources(XRAM=213, threads=1)
     future = create_dask_task(dask_task_impossible_resources)
     assert future
     assert await list_unrunnable_tasks(scheduler_url, scheduler_authentication) == [
         DaskTask(
             task_id=future.key,
-            required_resources=(
-                dask_task_impossible_resources | {DASK_WORKER_THREAD_RESOURCE_NAME: 1}
-            ),
+            required_resources=(dask_task_impossible_resources),
         )
     ]
     # remove that future, will remove the task
@@ -168,7 +168,7 @@ def _add_fct(x: int, y: int) -> int:
         next(iter(dask_spec_cluster_client.scheduler_info()["workers"])): [
             DaskTask(
                 task_id=DaskTaskId(future_queued_task.key),
-                required_resources={DASK_WORKER_THREAD_RESOURCE_NAME: 1},
+                required_resources=DaskTaskResources(threads=1),
             )
         ]
     }