create domain model for jobs

Author: bisgaard-itis

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/webserver/projects.py

@@ -81,15 +81,13 @@ async def get_project_marked_as_job(
     product_name: ProductName,
     user_id: UserID,
     project_uuid: ProjectID,
-    job_parent_resource_name: str,
 ) -> ProjectJobRpcGet:
     result = await rpc_client.request(
         WEBSERVER_RPC_NAMESPACE,
         TypeAdapter(RPCMethodName).validate_python("get_project_marked_as_job"),
         product_name=product_name,
         user_id=user_id,
         project_uuid=project_uuid,
-        job_parent_resource_name=job_parent_resource_name,
     )
     assert TypeAdapter(ProjectJobRpcGet).validate_python(result)  # nosec
     return cast(ProjectJobRpcGet, result)

services/api-server/src/simcore_service_api_server/_service_jobs.py

@@ -22,8 +22,11 @@
 from pydantic import HttpUrl
 from servicelib.logging_utils import log_context
 
+from .models.api_resources import RelativeResourceName
 from .models.basic_types import NameValueTuple
-from .models.schemas.jobs import Job, JobID, JobInputs
+from .models.domain.jobs import Job as DomainJob
+from .models.schemas.jobs import Job as SchemaJob
+from .models.schemas.jobs import JobID, JobInputs
 from .models.schemas.programs import Program
 from .models.schemas.solvers import Solver
 from .services_http.solver_job_models_converters import (
@@ -57,7 +60,7 @@ async def list_jobs(
         filter_any_custom_metadata: list[NameValueTuple] | None = None,
         pagination_offset: PageOffsetInt | None = None,
         pagination_limit: PageLimitInt | None = None,
-    ) -> tuple[list[Job], PageMetaInfoLimitOffset]:
+    ) -> tuple[list[SchemaJob], PageMetaInfoLimitOffset]:
         """Lists all jobs for a user with pagination based on resource name prefix"""
 
         pagination_kwargs = as_dict_exclude_none(
@@ -74,7 +77,7 @@ async def list_jobs(
         )
 
         # 2. Convert projects to jobs
-        jobs: list[Job] = []
+        jobs: list[SchemaJob] = []
         for project_job in projects_page.data:
             assert (  # nosec
                 len(project_job.workbench) == 1
@@ -87,9 +90,9 @@ async def list_jobs(
             assert project_job.job_parent_resource_name  # nosec
 
             jobs.append(
-                Job(
+                SchemaJob(
                     id=project_job.uuid,
-                    name=Job.compose_resource_name(
+                    name=SchemaJob.compose_resource_name(
                         project_job.job_parent_resource_name, project_job.uuid
                     ),
                     inputs_checksum=job_inputs.compute_checksum(),
@@ -114,12 +117,12 @@ async def create_job(
         hidden: bool,
         project_name: str | None,
         description: str | None,
-    ) -> tuple[Job, ProjectGet]:
+    ) -> tuple[SchemaJob, ProjectGet]:
         """If no project_name is provided, the job name is used as project name"""
 
         # creates NEW job as prototype
 
-        pre_job = Job.create_job_from_solver_or_program(
+        pre_job = SchemaJob.create_job_from_solver_or_program(
             solver_or_program_name=solver_or_program.name, inputs=inputs
         )
         with log_context(
@@ -155,7 +158,7 @@ async def create_job(
         )
         assert job.id == pre_job.id  # nosec
         assert job.name == pre_job.name  # nosec
-        assert job.name == Job.compose_resource_name(
+        assert job.name == SchemaJob.compose_resource_name(
             parent_name=solver_or_program.resource_name,
             job_id=job.id,
         )
@@ -175,8 +178,18 @@ async def start_log_export(
         )
         return async_job_get
 
+    async def get_job(
+        self,
+        job_id: JobID,
+    ) -> DomainJob:
+        return await self._web_rpc_client.get_project_marked_as_job(
+            product_name=self.product_name,
+            user_id=self.user_id,
+            project_id=job_id,
+        )
+
     async def delete_job_assets(
-        self, solver_or_program: Solver | Program, project_id: ProjectID
+        self, job_parent_resource_name: RelativeResourceName, project_id: ProjectID
     ):
         """Marks job project as hidden and deletes S3 assets associated it"""
         await self._web_rest_client.patch_project(
@@ -189,6 +202,6 @@ async def delete_job_assets(
             product_name=self.product_name,
             user_id=self.user_id,
             project_uuid=project_id,
-            job_parent_resource_name=solver_or_program.name,
+            job_parent_resource_name=job_parent_resource_name,
             storage_data_deleted=True,
         )

services/api-server/src/simcore_service_api_server/api/routes/solvers_jobs.py

@@ -14,7 +14,7 @@
 
 from ..._service_jobs import JobService
 from ..._service_solvers import SolverService
-from ...exceptions.backend_errors import ProjectAlreadyStartedError
+from ...exceptions.backend_errors import JobNotFoundError, ProjectAlreadyStartedError
 from ...exceptions.service_errors_utils import DEFAULT_BACKEND_SERVICE_STATUS_CODES
 from ...models.basic_types import VersionStr
 from ...models.schemas.errors import ErrorGet
@@ -167,14 +167,16 @@ async def delete_job_assets(
     version: VersionStr,
     job_id: JobID,
     job_service: Annotated[JobService, Depends(get_job_service)],
-    solver_service: Annotated[SolverService, Depends(get_solver_service)],
 ):
-    solver = await solver_service.get_solver(
-        solver_key=solver_key,
-        solver_version=version,
-    )
+    job = await job_service.get_job(job_id=job_id)
+    if job.job_parent_resource_name != compose_job_resource_name(
+        solver_key, version, job_id
+    ):
+        raise JobNotFoundError(
+            project_id=job_id,
+        )
     await job_service.delete_job_assets(
-        job_parent_resource_name=solver.name, project_id=job_id
+        job_parent_resource_name=job.job_parent_resource_name, project_id=job_id
     )
 
 

services/api-server/src/simcore_service_api_server/exceptions/backend_errors.py

@@ -57,6 +57,11 @@ class ServiceForbiddenAccessError(BaseBackEndError):
     status_code = status.HTTP_403_FORBIDDEN
 
 
+class JobForbiddenAccessError(BaseBackEndError):
+    msg_template = "Could not get solver/study job {project_id}"
+    status_code = status.HTTP_403_FORBIDDEN
+
+
 class JobNotFoundError(BaseBackEndError):
     msg_template = "Could not get solver/study job {project_id}"
     status_code = status.HTTP_404_NOT_FOUND

services/api-server/src/simcore_service_api_server/models/domain/jobs.py

@@ -0,0 +1,46 @@
+import datetime
+from typing import TypeAlias
+from uuid import UUID
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+)
+
+from ..api_resources import (
+    RelativeResourceName,
+)
+
+JobID: TypeAlias = UUID
+# JOB SUB-RESOURCES  ----------
+#
+#  - Wrappers for input/output values
+#  - Input/outputs are defined in service metadata
+#  - custom metadata
+#
+
+
+class Job(BaseModel):
+    id: JobID
+    name: RelativeResourceName
+    job_parent_resource_name: RelativeResourceName
+
+    created_at: datetime.datetime = Field(..., description="Job creation timestamp")
+
+    # parent
+    runner_name: RelativeResourceName = Field(
+        ..., description="Runner that executes job"
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "id": "f622946d-fd29-35b9-a193-abdd1095167c",
+                "name": "solvers/isolve/releases/1.3.4/jobs/f622946d-fd29-35b9-a193-abdd1095167c",
+                "runner_name": "solvers/isolve/releases/1.3.4",
+                "inputs_checksum": "12345",
+                "created_at": "2021-01-22T23:59:52.322176",
+            }
+        }
+    )

services/api-server/src/simcore_service_api_server/models/schemas/jobs.py

@@ -38,6 +38,8 @@
 from ..basic_types import VersionStr
 from ..domain.files import File as DomainFile
 from ..domain.files import FileInProgramJobData
+from ..domain.jobs import Job as DomainJob
+from ..domain.jobs import JobID
 from ..schemas.files import UserFile
 from .base import ApiServerInputSchema
 
@@ -50,8 +52,6 @@
 from .programs import Program, ProgramKeyId
 from .solvers import Solver
 
-JobID: TypeAlias = UUID
-
 # ArgumentTypes are types used in the job inputs (see ResultsTypes)
 ArgumentTypes: TypeAlias = (
     File | StrictFloat | StrictInt | StrictBool | str | list | None
@@ -324,6 +324,11 @@ def resource_name(self) -> str:
         return self.name
 
 
+assert set(DomainJob.model_fields.keys()).issubset(
+    set(Job.model_fields.keys())
+)  # nosec
+
+
 def get_url(
     solver_or_program: Solver | Program, url_for: Callable[..., HttpUrl], job_id: JobID
 ) -> HttpUrl | None:

services/api-server/src/simcore_service_api_server/services_rpc/wb_api_server.py

@@ -44,6 +44,7 @@
 from models_library.rpc.webserver.projects import (
     ListProjectsMarkedAsJobRpcFilters,
     MetadataFilterItem,
+    PageRpcProjectJobRpcGet,
 )
 from models_library.services_types import ServiceRunID
 from models_library.users import UserID
@@ -63,6 +64,10 @@
     NotEnoughAvailableSeatsError,
 )
 from servicelib.rabbitmq.rpc_interfaces.webserver import projects as projects_rpc
+from servicelib.rabbitmq.rpc_interfaces.webserver.errors import (
+    ProjectForbiddenRpcError,
+    ProjectNotFoundRpcError,
+)
 from servicelib.rabbitmq.rpc_interfaces.webserver.functions import (
     functions_rpc_interface,
 )
@@ -79,14 +84,21 @@
     release_licensed_item_for_wallet as _release_licensed_item_for_wallet,
 )
 from simcore_service_api_server.models.basic_types import NameValueTuple
+from simcore_service_api_server.models.domain.jobs import Job
 
 from ..exceptions.backend_errors import (
     CanNotCheckoutServiceIsNotRunningError,
     InsufficientNumberOfSeatsError,
+    JobForbiddenAccessError,
+    JobNotFoundError,
     LicensedItemCheckoutNotFoundError,
 )
 from ..exceptions.service_errors_utils import service_exception_mapper
-from ..models.api_resources import RelativeResourceName
+from ..models.api_resources import (
+    RelativeResourceName,
+    compose_resource_name,
+    split_resource_name,
+)
 from ..models.pagination import Page, PaginationParams
 from ..models.schemas.model_adapter import (
     LicensedItemCheckoutGet,
@@ -262,7 +274,7 @@ async def list_projects_marked_as_jobs(
         pagination_limit: int = 50,
         filter_by_job_parent_resource_name_prefix: str | None,
         filter_any_custom_metadata: list[NameValueTuple] | None,
-    ):
+    ) -> PageRpcProjectJobRpcGet:
         pagination_kwargs = as_dict_exclude_none(
             offset=pagination_offset, limit=pagination_limit
         )
@@ -287,6 +299,38 @@ async def list_projects_marked_as_jobs(
             **pagination_kwargs,
         )
 
+    @_exception_mapper(
+        rpc_exception_map={
+            ProjectForbiddenRpcError: JobForbiddenAccessError,
+            ProjectNotFoundRpcError: JobNotFoundError,
+        }
+    )
+    async def get_project_marked_as_job(
+        self,
+        *,
+        product_name: ProductName,
+        user_id: UserID,
+        project_id: ProjectID,
+    ) -> Job:
+        project_job_rpc = await projects_rpc.get_project_marked_as_job(
+            rpc_client=self._client,
+            product_name=product_name,
+            user_id=user_id,
+            project_uuid=project_id,
+        )
+        collection_or_resource_ids = [
+            *split_resource_name(project_job_rpc.job_parent_resource_name),
+            "jobs",
+            f"{project_job_rpc.uuid}",
+        ]
+        return Job(
+            id=project_job_rpc.uuid,
+            name=compose_resource_name(*collection_or_resource_ids),
+            job_parent_resource_name=project_job_rpc.job_parent_resource_name,
+            created_at=project_job_rpc.created_at,
+            runner_name=project_job_rpc.job_parent_resource_name,
+        )
+
     async def register_function(
         self, *, user_id: UserID, product_name: ProductName, function: Function
     ) -> RegisteredFunction: