@pcrespov decouple task models

Author: bisgaard-itis

packages/models-library/src/models_library/api_schemas_rpc_long_running_tasks/tasks.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+from typing import Any, TypeAlias
+from uuid import UUID
+
+from pydantic import BaseModel, Field, PositiveFloat, model_validator
+from typing_extensions import Self
+
+TaskRpcId: TypeAlias = UUID
+
+
+class TaskRpcStatus(BaseModel):
+    task_id: TaskRpcId
+    task_progress: PositiveFloat = Field(..., ge=0.0, le=1.0)
+    done: bool
+    started: datetime
+    stopped: datetime | None
+
+    @model_validator(mode="after")
+    def _check_consistency(self) -> Self:
+        progress_done = self.task_progress == 1.0
+        done_done = self.done == True
+        stopped_done = self.stopped is not None
+
+        if (progress_done != done_done) or (done_done != stopped_done):
+            raise ValueError(
+                f"Inconsistent data: {self.task_progress=}, {self.done=}, {self.stopped=}"
+            )
+        return self
+
+
+class TaskRpcResult(BaseModel):
+    result: Any | None
+    error: Any | None
+
+
+class TaskRpcGet(BaseModel):
+    task_id: TaskRpcId
+    task_name: str

packages/models-library/src/models_library/api_schemas_storage/zipping_tasks.py

@@ -1,7 +1,7 @@
 # pylint: disable=R6301
 from pathlib import Path
 
-from models_library.api_schemas_long_running_tasks.base import TaskId
+from models_library.api_schemas_rpc_long_running_tasks.tasks import TaskRpcId
 from pydantic import BaseModel, Field
 
 
@@ -11,4 +11,4 @@ class ZipTaskStartInput(BaseModel):
 
 class ZipTaskAbortOutput(BaseModel):
     result: bool
-    task_id: TaskId
+    task_id: TaskRpcId

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

@@ -1,10 +1,9 @@
 from typing import Final
 
-from models_library.api_schemas_long_running_tasks.base import TaskId
-from models_library.api_schemas_long_running_tasks.tasks import (
-    TaskGet,
-    TaskResult,
-    TaskStatus,
+from models_library.api_schemas_rpc_long_running_tasks.tasks import (
+    TaskRpcGet,
+    TaskRpcResult,
+    TaskRpcStatus,
 )
 from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
 from models_library.api_schemas_storage.zipping_tasks import (
@@ -13,6 +12,7 @@
 )
 from models_library.rabbitmq_basic_types import RPCMethodName
 from pydantic import NonNegativeInt, TypeAdapter
+from simcore_service_storage.api.rpc._zipping import TaskRpcId
 
 from ... import RabbitMQRPCClient
 
@@ -23,19 +23,19 @@
 
 async def start_zipping(
     rabbitmq_rpc_client: RabbitMQRPCClient, *, paths: ZipTaskStartInput
-) -> TaskGet:
+) -> TaskRpcGet:
     result = await rabbitmq_rpc_client.request(
         STORAGE_RPC_NAMESPACE,
         _RPC_METHOD_NAME_ADAPTER.validate_python("start_zipping"),
         paths=paths,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
-    assert isinstance(result, TaskGet)
+    assert isinstance(result, TaskRpcGet)
     return result
 
 
 async def abort_zipping(
-    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskId
+    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskRpcId
 ) -> ZipTaskAbortOutput:
     result = await rabbitmq_rpc_client.request(
         STORAGE_RPC_NAMESPACE,
@@ -48,26 +48,26 @@ async def abort_zipping(
 
 
 async def get_zipping_status(
-    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskId
-) -> TaskStatus:
+    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskRpcId
+) -> TaskRpcStatus:
     result = await rabbitmq_rpc_client.request(
         STORAGE_RPC_NAMESPACE,
         _RPC_METHOD_NAME_ADAPTER.validate_python("get_zipping_status"),
         task_id=task_id,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
-    assert isinstance(result, TaskStatus)
+    assert isinstance(result, TaskRpcStatus)
     return result
 
 
 async def get_zipping_result(
-    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskId
-) -> TaskResult:
+    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskRpcId
+) -> TaskRpcResult:
     result = await rabbitmq_rpc_client.request(
         STORAGE_RPC_NAMESPACE,
         _RPC_METHOD_NAME_ADAPTER.validate_python("get_zipping_result"),
         task_id=task_id,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
-    assert isinstance(result, TaskResult)
+    assert isinstance(result, TaskRpcResult)
     return result

services/storage/src/simcore_service_storage/api/rpc/_zipping.py

@@ -2,15 +2,11 @@
 from uuid import uuid4
 
 from fastapi import FastAPI
-from models_library.api_schemas_long_running_tasks.base import (
-    ProgressPercent,
-    TaskId,
-    TaskProgress,
-)
-from models_library.api_schemas_long_running_tasks.tasks import (
-    TaskGet,
-    TaskResult,
-    TaskStatus,
+from models_library.api_schemas_rpc_long_running_tasks.tasks import (
+    TaskRpcGet,
+    TaskRpcId,
+    TaskRpcResult,
+    TaskRpcStatus,
 )
 from models_library.api_schemas_storage.zipping_tasks import (
     ZipTaskAbortOutput,
@@ -22,36 +18,34 @@
 
 
 @router.expose()
-async def start_zipping(app: FastAPI, paths: ZipTaskStartInput) -> TaskGet:
+async def start_zipping(app: FastAPI, paths: ZipTaskStartInput) -> TaskRpcGet:
     assert app  # nosec
-    return TaskGet(
-        task_id=f"{uuid4()}",
+    return TaskRpcGet(
+        task_id=uuid4(),
         task_name=", ".join(str(p) for p in paths.paths),
-        status_href="status_url",
-        result_href="result url",
-        abort_href="abort url",
     )
 
 
 @router.expose()
-async def abort_zipping(app: FastAPI, task_id: TaskId) -> ZipTaskAbortOutput:
+async def abort_zipping(app: FastAPI, task_id: TaskRpcId) -> ZipTaskAbortOutput:
     assert app  # nosec
     return ZipTaskAbortOutput(result=True, task_id=task_id)
 
 
 @router.expose()
-async def get_zipping_status(app: FastAPI, task_id: TaskId) -> TaskStatus:
+async def get_zipping_status(app: FastAPI, task_id: TaskRpcId) -> TaskRpcStatus:
     assert app  # nosec
-    progress = TaskProgress(
+    return TaskRpcStatus(
         task_id=task_id,
-        message="Here's a status for you. You are welcome",
-        percent=ProgressPercent(0.5),
+        task_progress=0.5,
+        done=False,
+        started=datetime.now(),
+        stopped=None,
     )
-    return TaskStatus(task_progress=progress, done=False, started=datetime.now())
 
 
 @router.expose()
-async def get_zipping_result(app: FastAPI, task_id: TaskId) -> TaskResult:
+async def get_zipping_result(app: FastAPI, task_id: TaskRpcId) -> TaskRpcResult:
     assert app  # nosec
     assert task_id  # nosec
-    return TaskResult(result="Here's your result.", error=None)
+    return TaskRpcResult(result="Here's your result.", error=None)

services/storage/src/simcore_service_storage/core/application.py

@@ -25,8 +25,8 @@
     APP_NAME,
     APP_STARTED_BANNER_MSG,
 )
-from ..api.rabbitmq_rpc.routes import setup_rpc_api_routes
 from ..api.rest.routes import setup_rest_api_routes
+from ..api.rpc.routes import setup_rpc_api_routes
 from ..dsm import setup_dsm
 from ..dsm_cleaner import setup_dsm_cleaner
 from ..exceptions.handlers import set_exception_handlers

services/storage/tests/unit/test_zipping.py

@@ -6,7 +6,12 @@
 import pytest
 from faker import Faker
 from fastapi import FastAPI
-from models_library.api_schemas_long_running_tasks.tasks import TaskGet, TaskStatus
+from models_library.api_schemas_rpc_long_running_tasks.tasks import (
+    TaskRpcGet,
+    TaskRpcId,
+    TaskRpcResult,
+    TaskRpcStatus,
+)
 from models_library.api_schemas_storage.zipping_tasks import (
     ZipTaskAbortOutput,
     ZipTaskStartInput,
@@ -17,7 +22,6 @@
 from servicelib.rabbitmq import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.storage import zipping
 from settings_library.rabbit import RabbitSettings
-from simcore_service_storage.api.rabbitmq_rpc._zipping import TaskId, TaskResult
 from simcore_service_storage.core.settings import ApplicationSettings
 
 pytest_plugins = [
@@ -73,24 +77,24 @@ async def test_start_zipping(rpc_client: RabbitMQRPCClient, faker: Faker):
     result = await zipping.start_zipping(
         rpc_client, paths=ZipTaskStartInput(paths=[Path(faker.file_path())])
     )
-    assert isinstance(result, TaskGet)
+    assert isinstance(result, TaskRpcGet)
 
 
 async def test_abort_zipping(rpc_client: RabbitMQRPCClient, faker: Faker):
-    _task_id = TaskId(f"{faker.uuid4()}")
+    _task_id = TaskRpcId(faker.uuid4())
     result = await zipping.abort_zipping(rpc_client, task_id=_task_id)
     assert isinstance(result, ZipTaskAbortOutput)
     assert result.task_id == _task_id
 
 
 async def test_get_zipping_status(rpc_client: RabbitMQRPCClient, faker: Faker):
-    _task_id = TaskId(f"{faker.uuid4()}")
+    _task_id = TaskRpcId(faker.uuid4())
     result = await zipping.get_zipping_status(rpc_client, task_id=_task_id)
-    assert isinstance(result, TaskStatus)
-    assert result.task_progress.task_id == _task_id
+    assert isinstance(result, TaskRpcStatus)
+    assert result.task_id == _task_id
 
 
 async def test_get_zipping_result(rpc_client: RabbitMQRPCClient, faker: Faker):
-    _task_id = TaskId(f"{faker.uuid4()}")
+    _task_id = TaskRpcId(faker.uuid4())
     result = await zipping.get_zipping_result(rpc_client, task_id=_task_id)
-    assert isinstance(result, TaskResult)
+    assert isinstance(result, TaskRpcResult)