first attempt at definining run function task

Author: bisgaard-itis

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

@@ -4,7 +4,6 @@
 from common_library.json_serialization import json_dumps
 from cryptography.fernet import Fernet
 from fastapi import Depends, FastAPI, HTTPException, status
-from fastapi.requests import Request
 
 from ..._constants import MSG_BACKEND_SERVICE_UNAVAILABLE
 from ...core.settings import ApplicationSettings, WebServerSettings
@@ -29,19 +28,16 @@ def _get_settings(
     return settings
 
 
-def _get_encrypt(request: Request) -> Fernet | None:
-    e: Fernet | None = getattr(request.app.state, "webserver_fernet", None)
-    return e
-
-
 def get_session_cookie(
     identity: Annotated[str, Depends(get_active_user_email)],
     settings: Annotated[WebServerSettings, Depends(_get_settings)],
-    fernet: Annotated[Fernet | None, Depends(_get_encrypt)],
+    app: Annotated[FastAPI, Depends(get_app)],
 ) -> dict:
     # Based on aiohttp_session and aiohttp_security
     # SEE services/web/server/tests/unit/with_dbs/test_login.py
 
+    fernet: Fernet | None = getattr(app.state, "webserver_fernet", None)
+
     if fernet is None:
         raise HTTPException(
             status.HTTP_503_SERVICE_UNAVAILABLE, detail=MSG_BACKEND_SERVICE_UNAVAILABLE

services/api-server/src/simcore_service_api_server/celery/_worker_tasks/_functions_tasks.py

@@ -0,0 +1,108 @@
+from celery import Task
+from celery_library.utils import get_app_server
+from fastapi import FastAPI  # type: ignore[import-untyped]
+from models_library.functions import FunctionInputs, RegisteredFunction
+from models_library.projects_nodes_io import NodeID
+from servicelib.celery.models import TaskID
+
+from ...api.dependencies.authentication import Identity
+from ...api.dependencies.rabbitmq import get_rabbitmq_rpc_client
+from ...api.dependencies.services import (
+    get_api_client,
+    get_catalog_service,
+    get_directorv2_service,
+    get_function_job_service,
+    get_job_service,
+    get_solver_service,
+    get_storage_service,
+    get_wb_api_rpc_client,
+    get_webserver_session,
+)
+from ...api.dependencies.webserver_http import get_session_cookie
+from ...models.api_resources import JobLinks
+from ...models.schemas.jobs import JobPricingSpecification
+from ...services_http.director_v2 import DirectorV2Api
+from ...services_http.storage import StorageApi
+
+
+async def _assemble_function_job_service(app: FastAPI, identity: Identity):
+    # to avoid this show we could introduce a dependency injection
+    # system which is not linked to FastAPI (i.e. can be resolved manually).
+    # See also https://github.com/fastapi/fastapi/issues/1105#issuecomment-609919850
+    settings = app.state.settings
+    assert settings.API_SERVER_WEBSERVER  # nosec
+    session_cookie = get_session_cookie(
+        identity=identity.email, settings=settings.API_SERVER_WEBSERVER, app=app
+    )
+
+    rpc_client = get_rabbitmq_rpc_client(app=app)
+    web_server_rest_client = get_webserver_session(
+        app=app, session_cookies=session_cookie, identity=identity
+    )
+    web_api_rpc_client = await get_wb_api_rpc_client(app=app)
+    director2_api = get_api_client(DirectorV2Api)
+    assert isinstance(director2_api, DirectorV2Api)
+    storage_api = get_api_client(StorageApi)
+    assert isinstance(storage_api, StorageApi)
+    catalog_service = get_catalog_service(
+        rpc_client=rpc_client,
+        user_id=identity.user_id,
+        product_name=identity.product_name,
+    )
+
+    storage_service = get_storage_service(
+        rpc_client=rpc_client,
+        user_id=identity.user_id,
+        product_name=identity.product_name,
+    )
+    directorv2_service = get_directorv2_service(rpc_client=rpc_client)
+
+    solver_service = get_solver_service(
+        catalog_service=catalog_service,
+        user_id=identity.user_id,
+        product_name=identity.product_name,
+    )
+
+    job_service = get_job_service(
+        web_rest_api=web_server_rest_client,
+        director2_api=director2_api,
+        storage_api=storage_api,
+        web_rpc_api=web_api_rpc_client,
+        storage_service=storage_service,
+        directorv2_service=directorv2_service,
+        user_id=identity.user_id,
+        product_name=identity.product_name,
+        solver_service=solver_service,
+    )
+
+    return get_function_job_service(
+        web_rpc_api=web_api_rpc_client,
+        job_service=job_service,
+        user_id=identity.user_id,
+        product_name=identity.product_name,
+    )
+
+
+async def run_function(
+    task: Task,
+    task_id: TaskID,
+    *,
+    identity: Identity,  # user identity
+    function: RegisteredFunction,
+    function_inputs: FunctionInputs,
+    pricing_spec: JobPricingSpecification | None,
+    job_links: JobLinks,
+    x_simcore_parent_project_uuid: NodeID | None,
+    x_simcore_parent_node_id: NodeID | None,
+):
+    app = get_app_server(task.app).app
+    function_job_service = await _assemble_function_job_service(app, identity)
+
+    return await function_job_service.run_function(
+        function=function,
+        function_inputs=function_inputs,
+        pricing_spec=pricing_spec,
+        job_links=job_links,
+        x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
+        x_simcore_parent_node_id=x_simcore_parent_node_id,
+    )

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

@@ -13,7 +13,7 @@
 
 from ..core.application import create_app
 from ..core.settings import ApplicationSettings
-from ._worker_tasks import setup_worker_tasks
+from ._worker_tasks.tasks import setup_worker_tasks
 
 _settings = ApplicationSettings.create_from_envs()