purged unused

Author: Andrei Neagu

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

@@ -6,7 +6,6 @@
 from typing import Any, Literal, TypeAlias
 
 import arrow
-from models_library.osparc_jobs import OsparcJobId
 from pydantic import BaseModel, Field
 
 from .products import ProductName
@@ -51,11 +50,8 @@ def body(self) -> bytes:
         return self.model_dump_json().encode()
 
 
-class WorkerJobMessageBase(BaseModel):
+class ProjectMessageBase(BaseModel):
     user_id: UserID
-
-
-class ProjectMessageBase(WorkerJobMessageBase):
     project_id: ProjectID
 
 
@@ -97,13 +93,11 @@ class ProgressType(StrAutoEnum):
 
     PROJECT_CLOSING = auto()
 
-    WORKER_JOB_EXPORTING = auto()
-
 
 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
@@ -123,17 +117,10 @@ def routing_key(self) -> str | None:
         return f"{self.project_id}.all_nodes"
 
 
-class ProgressRabbitMessageWorkerJob(ProgressMessageMixin, WorkerJobMessageBase):
-    osparc_job_id: OsparcJobId
-
-    def routing_key(self) -> str | None:
-        return f"{self.user_id}.worker_job"
-
-
 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
@@ -223,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
@@ -263,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):
@@ -274,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(
         ...,
@@ -310,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/src/models_library/rabbitmq_messages.py

@@ -4,7 +4,6 @@
 from models_library.rabbitmq_messages import (
     ProgressRabbitMessageNode,
     ProgressRabbitMessageProject,
-    ProgressRabbitMessageWorkerJob,
     ProgressType,
 )
 from pydantic import TypeAdapter
@@ -36,22 +35,10 @@
             ProgressRabbitMessageProject,
             id="project_progress",
         ),
-        pytest.param(
-            ProgressRabbitMessageWorkerJob(
-                user_id=faker.pyint(min_value=1),
-                progress_type=ProgressType.PROJECT_CLOSING,
-                report=ProgressReport(actual_value=0.4, total=1),
-                osparc_job_id=faker.pystr(),
-            ).model_dump_json(),
-            ProgressRabbitMessageWorkerJob,
-            id="worker_job_progress",
-        ),
     ],
 )
 async def test_raw_message_parsing(raw_data: str, class_type: type):
     result = TypeAdapter(
-        ProgressRabbitMessageNode
-        | ProgressRabbitMessageProject
-        | ProgressRabbitMessageWorkerJob
+        ProgressRabbitMessageNode | ProgressRabbitMessageProject
     ).validate_json(raw_data)
     assert type(result) is class_type

packages/models-library/tests/test_rabbit_messages.py

@@ -7,7 +7,7 @@
 from models_library.users import UserID
 
 from ...dsm import get_dsm_provider
-from ...modules.celery.utils import get_celery_worker, get_fastapi_app
+from ...modules.celery.utils import get_celery_worker_client, get_fastapi_app
 from ...simcore_s3_dsm import SimcoreS3DataManager
 from ._tqdm_utils import get_export_progress, set_absolute_progress
 
@@ -33,7 +33,7 @@ async def data_export(
 
         async def _progress_cb(report: ProgressReport) -> None:
             set_absolute_progress(pbar, current_progress=report.actual_value)
-            await get_celery_worker(task.app).set_task_progress(task, report)
+            await get_celery_worker_client(task.app).set_task_progress(task, report)
 
         return await dsm.create_s3_export(
             user_id, paths_to_export, progress_cb=_progress_cb

services/storage/src/simcore_service_storage/api/_worker_tasks/_data_export.py

@@ -13,10 +13,10 @@
 from ...modules.celery import get_event_loop, set_event_loop
 from ...modules.celery.utils import (
     get_fastapi_app,
-    set_celery_worker,
+    set_celery_worker_client,
     set_fastapi_app,
 )
-from ...modules.celery.worker import CeleryTaskQueueWorker
+from ...modules.celery.worker import CeleryWorkerClient
 
 _logger = logging.getLogger(__name__)
 
@@ -48,7 +48,7 @@ async def lifespan():
         set_event_loop(fastapi_app, loop)
 
         set_fastapi_app(sender.app, fastapi_app)
-        set_celery_worker(sender.app, CeleryTaskQueueWorker(sender.app))
+        set_celery_worker_client(sender.app, CeleryWorkerClient(sender.app))
 
         loop.run_forever()
 

services/storage/src/simcore_service_storage/modules/celery/signals.py

@@ -1,15 +1,15 @@
 from celery import Celery  # type: ignore[import-untyped]
 from fastapi import FastAPI
 
-from .worker import CeleryTaskQueueWorker
+from .worker import CeleryWorkerClient
 
-_WORKER_KEY = "celery_worker"
+_WORKER_KEY = "celery_worker_client"
 _FASTAPI_APP_KEY = "fastapi_app"
 
 
-def get_celery_worker(celery_app: Celery) -> CeleryTaskQueueWorker:
+def get_celery_worker_client(celery_app: Celery) -> CeleryWorkerClient:
     worker = celery_app.conf[_WORKER_KEY]
-    assert isinstance(worker, CeleryTaskQueueWorker)
+    assert isinstance(worker, CeleryWorkerClient)
     return worker
 
 
@@ -19,7 +19,7 @@ def get_fastapi_app(celery_app: Celery) -> FastAPI:
     return fastapi_app
 
 
-def set_celery_worker(celery_app: Celery, worker: CeleryTaskQueueWorker) -> None:
+def set_celery_worker_client(celery_app: Celery, worker: CeleryWorkerClient) -> None:
     celery_app.conf[_WORKER_KEY] = worker
 
 

services/storage/src/simcore_service_storage/modules/celery/utils.py

@@ -8,7 +8,7 @@
 _logger = logging.getLogger(__name__)
 
 
-class CeleryTaskQueueWorker:
+class CeleryWorkerClient:
     def __init__(self, celery_app: Celery) -> None:
         self.celery_app = celery_app
 

services/storage/src/simcore_service_storage/modules/celery/worker.py

@@ -3,14 +3,7 @@
 from typing import cast
 
 from fastapi import FastAPI
-from models_library.osparc_jobs import OsparcJobId
-from models_library.progress_bar import ProgressReport
-from models_library.rabbitmq_messages import (
-    ProgressRabbitMessageWorkerJob,
-    ProgressType,
-    RabbitMessageBase,
-)
-from models_library.users import UserID
+from models_library.rabbitmq_messages import RabbitMessageBase
 from servicelib.logging_utils import log_catch, log_context
 from servicelib.rabbitmq import (
     RabbitMQClient,
@@ -77,16 +70,3 @@ async def post_message(app: FastAPI, message: RabbitMessageBase) -> None:
     with log_catch(_logger, reraise=False), suppress(ConfigurationError):
         # NOTE: if rabbitmq was not initialized the error does not need to flood the logs
         await get_rabbitmq_client(app).publish(message.channel_name, message)
-
-
-async def post_task_progress_message(
-    app: FastAPI, user_id: UserID, osparc_job_id: OsparcJobId, report: ProgressReport
-) -> None:
-    with log_catch(_logger, reraise=False):
-        message = ProgressRabbitMessageWorkerJob.model_construct(
-            user_id=user_id,
-            osparc_job_id=osparc_job_id,
-            progress_type=ProgressType.WORKER_JOB_EXPORTING,
-            report=report,
-        )
-        await post_message(app, message)

services/storage/src/simcore_service_storage/modules/rabbitmq.py

@@ -13,7 +13,7 @@
     on_worker_init,
     on_worker_shutdown,
 )
-from simcore_service_storage.modules.celery.worker import CeleryTaskQueueWorker
+from simcore_service_storage.modules.celery.worker import CeleryWorkerClient
 
 
 @pytest.fixture
@@ -97,6 +97,6 @@ def celery_worker_controller(
 @pytest.fixture
 def celery_worker(
     celery_worker_controller: TestWorkController,
-) -> CeleryTaskQueueWorker:
+) -> CeleryWorkerClient:
     assert isinstance(celery_worker_controller.app, Celery)
-    return CeleryTaskQueueWorker(celery_worker_controller.app)
+    return CeleryWorkerClient(celery_worker_controller.app)

services/storage/tests/unit/modules/celery/conftest.py

@@ -20,28 +20,24 @@
     TaskState,
 )
 from simcore_service_storage.modules.celery.utils import (
-    get_celery_worker,
+    get_celery_worker_client,
     get_fastapi_app,
 )
 from tenacity import Retrying, retry_if_exception_type, stop_after_delay, wait_fixed
 
 _logger = logging.getLogger(__name__)
 
 
-async def _async_archive(
-    celery_app: Celery, task_name: str, task_id: str, files: list[str]
-) -> str:
-    worker = get_celery_worker(celery_app)
+async def _async_archive(celery_app: Celery, task: Task, files: list[str]) -> str:
+    worker_client = get_celery_worker_client(celery_app)
 
     def sleep_for(seconds: float) -> None:
         time.sleep(seconds)
 
     for n, file in enumerate(files, start=1):
         with log_context(_logger, logging.INFO, msg=f"Processing file {file}"):
-            worker.set_task_progress(
-                task_name=task_name,
-                task_id=task_id,
-                report=ProgressReport(actual_value=n / len(files) * 10),
+            await worker_client.set_task_progress(
+                task, ProgressReport(actual_value=n / len(files) * 10)
             )
             await asyncio.get_event_loop().run_in_executor(None, sleep_for, 1)
 
@@ -52,7 +48,7 @@ def sync_archive(task: Task, files: list[str]) -> str:
     assert task.name
     _logger.info("Calling async_archive")
     return asyncio.run_coroutine_threadsafe(
-        _async_archive(task.app, task.name, task.request.id, files),
+        _async_archive(task, task.request.id, files),
         get_event_loop(get_fastapi_app(task.app)),
     ).result()