added endpoint for exporting function jobs logs

Author: bisgaard-itis

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

@@ -1,8 +1,10 @@
 import logging
 from collections.abc import Callable
 from dataclasses import dataclass
+from pathlib import Path
 
 from common_library.exclude import as_dict_exclude_none
+from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobGet
 from models_library.api_schemas_webserver.projects import ProjectCreateNew, ProjectGet
 from models_library.products import ProductName
 from models_library.projects import ProjectID
@@ -15,9 +17,9 @@
 from models_library.users import UserID
 from pydantic import HttpUrl
 from servicelib.logging_utils import log_context
-from simcore_service_api_server.models.basic_types import NameValueTuple
 
-from .models.schemas.jobs import Job, JobInputs
+from .models.basic_types import NameValueTuple
+from .models.schemas.jobs import Job, JobID, JobInputs
 from .models.schemas.programs import Program
 from .models.schemas.solvers import Solver
 from .services_http.solver_job_models_converters import (
@@ -26,6 +28,8 @@
     create_new_project_for_job,
 )
 from .services_http.webserver import AuthSession
+from .services_rpc.director_v2 import DirectorV2Service
+from .services_rpc.storage import StorageService
 from .services_rpc.wb_api_server import WbApiRpcClient
 
 _logger = logging.getLogger(__name__)
@@ -35,6 +39,8 @@
 class JobService:
     _web_rest_client: AuthSession
     _web_rpc_client: WbApiRpcClient
+    _storage_rpc_client: StorageService
+    _directorv2_rpc_client: DirectorV2Service
     user_id: UserID
     product_name: ProductName
 
@@ -147,3 +153,17 @@ async def create_job(
             job_id=job.id,
         )
         return job, new_project
+
+    async def start_log_export(
+        self,
+        job_id: JobID,
+    ) -> AsyncJobGet:
+        file_ids = await self._directorv2_rpc_client.get_computation_task_log_file_ids(
+            project_id=job_id
+        )
+        async_job_get = await self._storage_rpc_client.start_data_export(
+            paths_to_export=[
+                Path(elm.file_id) for elm in file_ids if elm.file_id is not None
+            ],
+        )
+        return async_job_get

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

@@ -14,6 +14,8 @@
 from ..._service_studies import StudyService
 from ...services_http.webserver import AuthSession
 from ...services_rpc.catalog import CatalogService
+from ...services_rpc.director_v2 import DirectorV2Service
+from ...services_rpc.storage import StorageService
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ...utils.client_base import BaseServiceClientApi
 from .authentication import get_current_user_id, get_product_name
@@ -61,9 +63,29 @@ def get_catalog_service(
     )
 
 
+def get_storage_service(
+    rpc_client: Annotated[RabbitMQRPCClient, Depends(get_rabbitmq_rpc_client)],
+    user_id: Annotated[UserID, Depends(get_current_user_id)],
+    product_name: Annotated[ProductName, Depends(get_product_name)],
+) -> StorageService:
+    return StorageService(
+        _rpc_client=rpc_client,
+        _user_id=user_id,
+        _product_name=product_name,
+    )
+
+
+def get_directorv2_service(
+    rpc_client: Annotated[RabbitMQRPCClient, Depends(get_rabbitmq_rpc_client)],
+) -> DirectorV2Service:
+    return DirectorV2Service(_rpc_client=rpc_client)
+
+
 def get_job_service(
     web_rest_api: Annotated[AuthSession, Depends(get_webserver_session)],
     web_rpc_api: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    storage_service: Annotated[StorageService, Depends(get_storage_service)],
+    directorv2_service: Annotated[DirectorV2Service, Depends(get_directorv2_service)],
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
 ) -> JobService:
@@ -74,6 +96,8 @@ def get_job_service(
     return JobService(
         _web_rest_client=web_rest_api,
         _web_rpc_client=web_rpc_api,
+        _storage_rpc_client=storage_service,
+        _directorv2_rpc_client=directorv2_service,
         user_id=user_id,
         product_name=product_name,
     )

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

@@ -1,7 +1,8 @@
 from typing import Annotated, Final
 
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends, FastAPI, status
 from fastapi_pagination.api import create_page
+from models_library.api_schemas_long_running_tasks.tasks import TaskGet
 from models_library.api_schemas_webserver.functions import (
     Function,
     FunctionClass,
@@ -17,8 +18,10 @@
 )
 from models_library.products import ProductName
 from models_library.users import UserID
+from servicelib.fastapi.dependencies import get_app
 from sqlalchemy.ext.asyncio import AsyncEngine
 
+from ..._service_jobs import JobService
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
 from ...services_http.director_v2 import DirectorV2Api
@@ -27,7 +30,7 @@
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ..dependencies.authentication import get_current_user_id, get_product_name
 from ..dependencies.database import get_db_asyncpg_engine
-from ..dependencies.services import get_api_client
+from ..dependencies.services import get_api_client, get_job_service
 from ..dependencies.webserver_http import get_webserver_session
 from ..dependencies.webserver_rpc import get_wb_api_rpc_client
 from . import solvers_jobs, solvers_jobs_read, studies_jobs
@@ -297,3 +300,58 @@ async def function_job_outputs(
             ).results
         )
     raise UnsupportedFunctionClassError(function_class=function.function_class)
+
+
+@function_job_router.post(
+    "/{function_job_id:uuid}/logs",
+    response_model=TaskGet,
+    responses={**_COMMON_FUNCTION_JOB_ERROR_RESPONSES},
+)
+async def start_function_job_logs(
+    function_job_id: FunctionJobID,
+    app: Annotated[FastAPI, Depends(get_app)],
+    job_service: Annotated[JobService, Depends(get_job_service)],
+    user_id: Annotated[UserID, Depends(get_current_user_id)],
+    wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    product_name: Annotated[ProductName, Depends(get_product_name)],
+):
+    function, function_job = await get_function_from_functionjobid(
+        wb_api_rpc=wb_api_rpc,
+        function_job_id=function_job_id,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    app_router = app.router
+
+    if (
+        function.function_class == FunctionClass.PROJECT
+        and function_job.function_class == FunctionClass.PROJECT
+    ):
+        async_job_get = await job_service.start_log_export(
+            job_id=function_job.project_job_id,
+        )
+        _task_id = f"{async_job_get.job_id}"
+        return TaskGet(
+            task_id=_task_id,
+            task_name=async_job_get.job_name,
+            status_href=app_router.url_path_for("get_task_status", task_id=_task_id),
+            abort_href=app_router.url_path_for("cancel_task", task_id=_task_id),
+            result_href=app_router.url_path_for("get_task_result", task_id=_task_id),
+        )
+
+    if (
+        function.function_class == FunctionClass.SOLVER
+        and function_job.function_class == FunctionClass.SOLVER
+    ):
+        async_job_get = await job_service.start_log_export(
+            job_id=function_job.solver_job_id,
+        )
+        _task_id = f"{async_job_get.job_id}"
+        return TaskGet(
+            task_id=_task_id,
+            task_name=async_job_get.job_name,
+            status_href=app_router.url_path_for("get_task_status", task_id=_task_id),
+            abort_href=app_router.url_path_for("cancel_task", task_id=_task_id),
+            result_href=app_router.url_path_for("get_task_result", task_id=_task_id),
+        )
+    raise UnsupportedFunctionClassError(function_class=function.function_class)

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

@@ -16,18 +16,18 @@
 @dataclass(frozen=True, kw_only=True)
 class StorageService:
     _rpc_client: RabbitMQRPCClient
+    _user_id: UserID
+    _product_name: ProductName
 
     @_exception_mapper(rpc_exception_map={})
     async def start_data_export(
         self,
-        user_id: UserID,
-        product_name: ProductName,
         paths_to_export: list[PathToExport],
     ) -> AsyncJobGet:
         async_job_get, _ = await start_export_data(
             self._rpc_client,
-            user_id=user_id,
-            product_name=product_name,
+            user_id=self._user_id,
+            product_name=self._product_name,
             paths_to_export=paths_to_export,
         )
         return async_job_get