transform run function endpoint

Author: bisgaard-itis

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

@@ -1,3 +1,5 @@
+from typing import Final
+
 from celery_library.common import create_app, create_task_manager
 from celery_library.task_manager import CeleryTaskManager
 from celery_library.types import register_celery_types, register_pydantic_types
@@ -13,6 +15,8 @@
 from ...models.api_resources import JobLinks
 from ...models.schemas.jobs import JobPricingSpecification
 
+ASYNC_JOB_CLIENT_NAME: Final[str] = "API_SERVER"
+
 
 def setup_task_manager(app: FastAPI, celery_settings: CelerySettings) -> None:
     async def on_startup() -> None:

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

@@ -2,6 +2,7 @@
 from collections.abc import Callable
 from typing import Annotated, Final, Literal
 
+from celery_library.task_manager import CeleryTaskManager
 from fastapi import APIRouter, Depends, Header, Request, status
 from fastapi_pagination.api import create_page
 from fastapi_pagination.bases import AbstractPage
@@ -16,19 +17,29 @@
     RegisteredFunctionJob,
     RegisteredFunctionJobCollection,
 )
+from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
 from models_library.products import ProductName
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
+from servicelib.celery.models import TaskFilter, TaskMetadata
 from servicelib.fastapi.dependencies import get_reverse_url_mapper
+from servicelib.long_running_tasks.models import TaskGet
 
 from ..._service_function_jobs import FunctionJobService
 from ..._service_functions import FunctionService
+from ...celery._worker_tasks._functions_tasks import run_function
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
 from ...models.schemas.jobs import JobPricingSpecification
 from ...services_rpc.wb_api_server import WbApiRpcClient
-from ..dependencies.authentication import get_current_user_id, get_product_name
+from ..dependencies.authentication import (
+    Identity,
+    get_current_identity,
+    get_current_user_id,
+    get_product_name,
+)
+from ..dependencies.celery import ASYNC_JOB_CLIENT_NAME, get_task_manager_from_app
 from ..dependencies.services import (
     get_function_job_service,
     get_function_service,
@@ -304,7 +315,7 @@ async def validate_function_inputs(
 
 @function_router.post(
     "/{function_id:uuid}:run",
-    response_model=RegisteredFunctionJob,
+    response_model=TaskGet,
     responses={**_COMMON_FUNCTION_ERROR_RESPONSES},
     description=create_route_description(
         base="Run function",
@@ -313,16 +324,15 @@ async def validate_function_inputs(
 )
 async def run_function(  # noqa: PLR0913
     request: Request,
+    user_identity: Annotated[Identity, Depends(get_current_identity)],
     to_run_function: Annotated[RegisteredFunction, Depends(get_function)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
     function_inputs: FunctionInputs,
     function_service: Annotated[FunctionService, Depends(get_function_service)],
-    function_jobs_service: Annotated[
-        FunctionJobService, Depends(get_function_job_service)
-    ],
+    task_manager: Annotated[CeleryTaskManager, Depends(get_task_manager_from_app)],
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
-) -> RegisteredFunctionJob:
+) -> TaskGet:
     parent_project_uuid = (
         x_simcore_parent_project_uuid
         if isinstance(x_simcore_parent_project_uuid, ProjectID)
@@ -336,7 +346,20 @@ async def run_function(  # noqa: PLR0913
     pricing_spec = JobPricingSpecification.create_from_headers(request.headers)
     job_links = await function_service.get_function_job_links(to_run_function, url_for)
 
-    return await function_jobs_service.run_function(
+    job_filter = AsyncJobFilter(
+        user_id=user_identity.user_id,
+        product_name=user_identity.product_name,
+        client_name=ASYNC_JOB_CLIENT_NAME,
+    )
+    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_name = run_function.__name__
+
+    task_uuid = await task_manager.submit_task(
+        TaskMetadata(
+            name=task_name,
+        ),
+        task_filter=task_filter,
+        user_identity=user_identity,
         function=to_run_function,
         function_inputs=function_inputs,
         pricing_spec=pricing_spec,
@@ -345,6 +368,14 @@ async def run_function(  # noqa: PLR0913
         x_simcore_parent_node_id=parent_node_id,
     )
 
+    return TaskGet(
+        task_id=f"{task_uuid}",
+        task_name=task_name,
+        status_href=url_for("get_task_status", task_id=task_uuid),
+        result_href=url_for("get_task_result", task_id=task_uuid),
+        abort_href=url_for("cancel_task", task_id=task_uuid),
+    )
+
 
 @function_router.delete(
     "/{function_id:uuid}",

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

@@ -1,5 +1,5 @@
 import logging
-from typing import Annotated, Any, Final
+from typing import Annotated, Any
 
 from fastapi import APIRouter, Depends, FastAPI, status
 from models_library.api_schemas_long_running_tasks.base import TaskProgress
@@ -20,21 +20,20 @@
 from ...models.schemas.errors import ErrorGet
 from ...services_rpc.async_jobs import AsyncJobClient
 from ..dependencies.authentication import get_current_user_id, get_product_name
+from ..dependencies.celery import ASYNC_JOB_CLIENT_NAME
 from ..dependencies.tasks import get_async_jobs_client
 from ._constants import (
     FMSG_CHANGELOG_NEW_IN_VERSION,
     create_route_description,
 )
 
-_ASYNC_JOB_CLIENT_NAME: Final[str] = "API_SERVER"
-
 router = APIRouter()
 _logger = logging.getLogger(__name__)
 
 
 def _get_job_filter(user_id: UserID, product_name: ProductName) -> AsyncJobFilter:
     return AsyncJobFilter(
-        user_id=user_id, product_name=product_name, client_name=_ASYNC_JOB_CLIENT_NAME
+        user_id=user_id, product_name=product_name, client_name=ASYNC_JOB_CLIENT_NAME
     )