port start_study_job code to job_service

Author: bisgaard-itis

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

@@ -43,7 +43,7 @@
 )
 from .models.schemas.programs import Program
 from .models.schemas.solvers import Solver, SolverKeyId
-from .models.schemas.studies import StudyID
+from .models.schemas.studies import Study, StudyID
 from .services_http.director_v2 import DirectorV2Api
 from .services_http.jobs import start_project
 from .services_http.solver_job_models_converters import (
@@ -65,14 +65,22 @@
 _logger = logging.getLogger(__name__)
 
 
-def compose_job_resource_name(solver_key, solver_version, job_id) -> str:
+def compose_solver_job_resource_name(solver_key, solver_version, job_id) -> str:
     """Creates a unique resource name for solver's jobs"""
     return Job.compose_resource_name(
         parent_name=Solver.compose_resource_name(solver_key, solver_version),
         job_id=job_id,
     )
 
 
+def compose_study_job_resource_name(study_key, job_id) -> str:
+    """Creates a unique resource name for study's jobs"""
+    return Job.compose_resource_name(
+        parent_name=Study.compose_resource_name(study_key),
+        job_id=job_id,
+    )
+
+
 @dataclass(frozen=True, kw_only=True)
 class JobService:
     _web_rest_client: AuthSession
@@ -352,7 +360,7 @@ async def start_solver_job(
         """
         Raises ProjectAlreadyStartedError if the project is already started
         """
-        job_name = compose_job_resource_name(solver_key, version, job_id)
+        job_name = compose_solver_job_resource_name(solver_key, version, job_id)
         _logger.debug("Start Job '%s'", job_name)
         job_parent_resource_name = Solver.compose_resource_name(solver_key, version)
         job = await self.get_job(
@@ -446,3 +454,21 @@ async def inspect_study_job(self, *, job_id: JobID) -> JobStatus:
         )
         job_status: JobStatus = create_jobstatus_from_task(task)
         return job_status
+
+    async def start_study_job(
+        self,
+        *,
+        job_id: JobID,
+        study_id: StudyID,
+        pricing_spec: JobPricingSpecification | None,
+    ):
+        job_name = compose_study_job_resource_name(study_id, job_id)
+        await start_project(
+            job_id=job_id,
+            expected_job_name=job_name,
+            webserver_api=self._web_rest_client,
+            pricing_spec=pricing_spec,
+        )
+        return await self.inspect_study_job(
+            job_id=job_id,
+        )

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

@@ -470,7 +470,6 @@ async def run_function(  # noqa: PLR0913
             request=request,
             study_id=to_run_function.project_id,
             job_id=study_job.id,
-            webserver_api=webserver_api,
             job_service=job_service,
         )
         return await register_function_job(

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

@@ -11,7 +11,7 @@
 from models_library.projects_nodes_io import NodeID
 from pydantic.types import PositiveInt
 
-from ..._service_jobs import JobService, compose_job_resource_name
+from ..._service_jobs import JobService, compose_solver_job_resource_name
 from ...exceptions.backend_errors import ProjectAlreadyStartedError
 from ...exceptions.service_errors_utils import DEFAULT_BACKEND_SERVICE_STATUS_CODES
 from ...models.basic_types import VersionStr
@@ -124,7 +124,7 @@ async def delete_job(
     job_id: JobID,
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
 ):
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Deleting Job '%s'", job_name)
 
     await webserver_api.delete_project(project_id=job_id)
@@ -244,7 +244,7 @@ async def stop_job(
     user_id: Annotated[PositiveInt, Depends(get_current_user_id)],
     director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
 ):
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Stopping Job '%s'", job_name)
 
     return await stop_project(
@@ -269,7 +269,7 @@ async def inspect_job(
     job_id: JobID,
     job_service: Annotated[JobService, Depends(get_job_service)],
 ) -> JobStatus:
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Inspecting Job '%s'", job_name)
 
     return await job_service.inspect_solver_job(
@@ -296,7 +296,7 @@ async def replace_job_custom_metadata(
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
 ):
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Custom metadata for '%s'", job_name)
 
     return await replace_custom_metadata(

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

@@ -21,7 +21,7 @@
 from sqlalchemy.ext.asyncio import AsyncEngine
 from starlette.background import BackgroundTask
 
-from ..._service_jobs import JobService, compose_job_resource_name
+from ..._service_jobs import JobService, compose_solver_job_resource_name
 from ..._service_solvers import SolverService
 from ...exceptions.custom_errors import InsufficientCreditsError, MissingWalletError
 from ...exceptions.service_errors_utils import DEFAULT_BACKEND_SERVICE_STATUS_CODES
@@ -275,7 +275,8 @@ async def get_job(
 ):
     """Gets job of a given solver"""
     _logger.debug(
-        "Getting Job '%s'", compose_job_resource_name(solver_key, version, job_id)
+        "Getting Job '%s'",
+        compose_solver_job_resource_name(solver_key, version, job_id),
     )
 
     solver = await solver_service.get_solver(
@@ -312,7 +313,7 @@ async def get_job_outputs(
     job_service: Annotated[JobService, Depends(get_job_service)],
     storage_client: Annotated[StorageApi, Depends(get_api_client(StorageApi))],
 ):
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Get Job '%s' outputs", job_name)
 
     project_marked_as_job = await job_service.get_job(
@@ -395,7 +396,7 @@ async def get_job_output_logfile(
     user_id: Annotated[PositiveInt, Depends(get_current_user_id)],
     director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
 ):
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Get Job '%s' outputs logfile", job_name)
 
     project_id = job_id
@@ -451,7 +452,7 @@ async def get_job_custom_metadata(
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
 ):
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Custom metadata for '%s'", job_name)
 
     return await get_custom_metadata(
@@ -481,7 +482,7 @@ async def get_job_wallet(
     job_id: JobID,
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
 ) -> WalletGetWithAvailableCreditsLegacy:
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     _logger.debug("Getting wallet for job '%s'", job_name)
 
     if project_wallet := await webserver_api.get_project_wallet(project_id=job_id):
@@ -504,7 +505,7 @@ async def get_job_pricing_unit(
     job_id: JobID,
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
 ):
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     with log_context(_logger, logging.DEBUG, "Get pricing unit"):
         _logger.debug("job: %s", job_name)
         project: ProjectGet = await webserver_api.get_project(project_id=job_id)
@@ -535,7 +536,7 @@ async def get_log_stream(
 ):
     assert request  # nosec
 
-    job_name = compose_job_resource_name(solver_key, version, job_id)
+    job_name = compose_solver_job_resource_name(solver_key, version, job_id)
     with log_context(
         _logger, logging.DEBUG, f"Streaming logs for {job_name=} and {user_id=}"
     ):

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

@@ -12,7 +12,7 @@
 from pydantic import HttpUrl, PositiveInt
 from servicelib.logging_utils import log_context
 
-from ..._service_jobs import JobService
+from ..._service_jobs import JobService, compose_study_job_resource_name
 from ...exceptions.backend_errors import ProjectAlreadyStartedError
 from ...models.api_resources import parse_resources_ids
 from ...models.pagination import Page, PaginationParams
@@ -27,12 +27,11 @@
     JobPricingSpecification,
     JobStatus,
 )
-from ...models.schemas.studies import JobLogsMap, Study, StudyID
+from ...models.schemas.studies import JobLogsMap, StudyID
 from ...services_http.director_v2 import DirectorV2Api
 from ...services_http.jobs import (
     get_custom_metadata,
     replace_custom_metadata,
-    start_project,
     stop_project,
 )
 from ...services_http.storage import StorageApi
@@ -57,14 +56,6 @@
 router = APIRouter()
 
 
-def _compose_job_resource_name(study_key, job_id) -> str:
-    """Creates a unique resource name for solver's jobs"""
-    return Job.compose_resource_name(
-        parent_name=Study.compose_resource_name(study_key),
-        job_id=job_id,
-    )
-
-
 @router.get(
     "/{study_id:uuid}/jobs",
     response_model=Page[Job],
@@ -129,7 +120,7 @@ async def create_study_job(
         x_simcore_parent_node_id=x_simcore_parent_node_id,
         hidden=hidden,
     )
-    assert job.name == _compose_job_resource_name(study_id, job.id)
+    assert job.name == compose_study_job_resource_name(study_id, job.id)
     job.url = url_for(
         "get_study_job",
         study_id=study_id,
@@ -175,7 +166,7 @@ async def delete_study_job(
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
 ):
     """Deletes an existing study job"""
-    job_name = _compose_job_resource_name(study_id, job_id)
+    job_name = compose_study_job_resource_name(study_id, job_id)
     with log_context(_logger, logging.DEBUG, f"Deleting Job '{job_name}'"):
         await webserver_api.delete_project(project_id=job_id)
 
@@ -212,7 +203,6 @@ async def start_study_job(
     request: Request,
     study_id: StudyID,
     job_id: JobID,
-    webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     job_service: Annotated[JobService, Depends(get_job_service)],
     cluster_id: Annotated[  # pylint: disable=unused-argument  # noqa: ARG001
         ClusterID | None,
@@ -230,29 +220,21 @@ async def start_study_job(
 ):
     pricing_spec = JobPricingSpecification.create_from_headers(headers=request.headers)
 
-    job_name = _compose_job_resource_name(study_id, job_id)
+    job_name = compose_study_job_resource_name(study_id, job_id)
     with log_context(_logger, logging.DEBUG, f"Starting Job '{job_name}'"):
         try:
-            await start_project(
+            return await job_service.start_study_job(
+                study_id=study_id,
                 job_id=job_id,
-                expected_job_name=job_name,
-                webserver_api=webserver_api,
                 pricing_spec=pricing_spec,
             )
         except ProjectAlreadyStartedError:
-            job_status: JobStatus = await inspect_study_job(
-                study_id=study_id,
+            job_status: JobStatus = await job_service.inspect_study_job(
                 job_id=job_id,
-                job_service=job_service,
             )
             return JSONResponse(
                 content=jsonable_encoder(job_status), status_code=status.HTTP_200_OK
             )
-        return await inspect_study_job(
-            study_id=study_id,
-            job_id=job_id,
-            job_service=job_service,
-        )
 
 
 @router.post(
@@ -265,7 +247,7 @@ async def stop_study_job(
     user_id: Annotated[PositiveInt, Depends(get_current_user_id)],
     director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
 ):
-    job_name = _compose_job_resource_name(study_id, job_id)
+    job_name = compose_study_job_resource_name(study_id, job_id)
     with log_context(_logger, logging.DEBUG, f"Stopping Job '{job_name}'"):
         return await stop_project(
             job_id=job_id, user_id=user_id, director2_api=director2_api
@@ -281,7 +263,7 @@ async def inspect_study_job(
     job_id: JobID,
     job_service: Annotated[JobService, Depends(get_job_service)],
 ) -> JobStatus:
-    job_name = _compose_job_resource_name(study_id, job_id)
+    job_name = compose_study_job_resource_name(study_id, job_id)
     _logger.debug("Inspecting Job '%s'", job_name)
 
     return await job_service.inspect_study_job(job_id=job_id)
@@ -298,7 +280,7 @@ async def get_study_job_outputs(
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     storage_client: Annotated[StorageApi, Depends(get_api_client(StorageApi))],
 ):
-    job_name = _compose_job_resource_name(study_id, job_id)
+    job_name = compose_study_job_resource_name(study_id, job_id)
     _logger.debug("Getting Job Outputs for '%s'", job_name)
 
     project_outputs = await webserver_api.get_project_outputs(project_id=job_id)
@@ -345,7 +327,7 @@ async def get_study_job_custom_metadata(
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
 ):
-    job_name = _compose_job_resource_name(study_id, job_id)
+    job_name = compose_study_job_resource_name(study_id, job_id)
     msg = f"Gets metadata attached to study_id={study_id!r} job_id={job_id!r}.\njob_name={job_name!r}.\nSEE https://github.com/ITISFoundation/osparc-simcore/issues/4313"
     _logger.debug(msg)
 
@@ -376,7 +358,7 @@ async def replace_study_job_custom_metadata(
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
 ):
-    job_name = _compose_job_resource_name(study_id, job_id)
+    job_name = compose_study_job_resource_name(study_id, job_id)
 
     msg = f"Attaches metadata={replace.metadata!r} to study_id={study_id!r} job_id={job_id!r}.\njob_name={job_name!r}.\nSEE https://github.com/ITISFoundation/osparc-simcore/issues/4313"
     _logger.debug(msg)