fix: remove rpc

Author: giancarloromeo

packages/service-library/src/servicelib/celery/tasks/__init__.py

@@ -0,0 +1,7 @@
+from ..models import ExecutionMetadata
+
+SEARCH_TASK_NAME = "search"
+SEARCH_EXECUTION_METADATA = ExecutionMetadata(
+    name=SEARCH_TASK_NAME,
+    streamed_result=True,
+)

packages/service-library/src/servicelib/celery/tasks/storage.py

@@ -9,6 +9,7 @@
     FoldersBody,
     PresignedLink,
 )
+from servicelib.celery.tasks.storage import SEARCH_TASK_NAME
 from servicelib.logging_utils import log_context
 
 from ...models import FileMetaData
@@ -41,4 +42,4 @@ def setup_worker_tasks(app: Celery) -> None:
         register_task(app, complete_upload_file)
         register_task(app, delete_paths)
         register_task(app, deep_copy_files_from_project)
-        register_task(app, search)
+        register_task(app, search, task_name=SEARCH_TASK_NAME)

services/storage/src/simcore_service_storage/api/_worker_tasks/tasks.py

@@ -1,12 +1,10 @@
-import datetime
 from typing import Literal
 
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
 )
 from models_library.api_schemas_storage.storage_schemas import FoldersBody
 from models_library.api_schemas_webserver.storage import PathToExport
-from models_library.projects import ProjectID
 from models_library.users import UserID
 from servicelib.celery.models import (
     ExecutionMetadata,
@@ -20,7 +18,6 @@
     deep_copy_files_from_project,
     export_data,
     export_data_as_download_link,
-    search,
 )
 
 router = RPCRouter()
@@ -71,31 +68,3 @@ async def start_export_data(
         paths_to_export=paths_to_export,
     )
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
-
-
-@router.expose()
-async def start_search(
-    task_manager: TaskManager,
-    owner_metadata: OwnerMetadata,
-    user_id: UserID,
-    limit: int,
-    name_pattern: str,
-    modified_at: (
-        tuple[datetime.datetime | None, datetime.datetime | None] | None
-    ) = None,
-    project_id: ProjectID | None = None,
-) -> AsyncJobGet:
-    task_name = search.__name__
-    task_uuid = await task_manager.submit_task(
-        execution_metadata=ExecutionMetadata(
-            name=task_name,
-            streamed_result=True,
-        ),
-        owner_metadata=owner_metadata,
-        user_id=user_id,
-        project_id=project_id,
-        limit=limit,
-        name_pattern=name_pattern,
-        modified_at=modified_at,
-    )
-    return AsyncJobGet(job_id=task_uuid, job_name=task_name)

services/storage/src/simcore_service_storage/api/rpc/_simcore_s3.py

@@ -48,21 +48,20 @@
 )
 from servicelib.aiohttp.rest_responses import create_data_response
 from servicelib.celery.models import OwnerMetadata
+from servicelib.celery.tasks.storage import SEARCH_EXECUTION_METADATA, SEARCH_TASK_NAME
 from servicelib.common_headers import X_FORWARDED_PROTO
 from servicelib.rabbitmq.rpc_interfaces.storage.paths import (
     compute_path_size as remote_compute_path_size,
 )
 from servicelib.rabbitmq.rpc_interfaces.storage.paths import (
     delete_paths as remote_delete_paths,
 )
-from servicelib.rabbitmq.rpc_interfaces.storage.simcore_s3 import (
-    start_export_data,
-    start_search,
-)
+from servicelib.rabbitmq.rpc_interfaces.storage.simcore_s3 import start_export_data
 from servicelib.rest_responses import unwrap_envelope
 from yarl import URL
 
 from .._meta import API_VTAG
+from ..celery import get_task_manager
 from ..login.decorators import login_required
 from ..models import AuthenticatedRequestContext, WebServerOwnerMetadata
 from ..rabbitmq import get_rabbitmq_rpc_client
@@ -549,15 +548,14 @@ async def search(request: web.Request) -> web.Response:
     class _PathParams(BaseModel):
         location_id: Annotated[LocationID, AfterValidator(_allow_only_simcore)]
 
-    rabbitmq_rpc_client = get_rabbitmq_rpc_client(request.app)
     _req_ctx = AuthenticatedRequestContext.model_validate(request)
     parse_request_path_parameters_as(_PathParams, request)
     search_body = await parse_request_body_as(
         model_schema_cls=SearchBodyParams, request=request
     )
 
-    async_job_rpc_get, _ = await start_search(
-        rabbitmq_rpc_client,
+    task_uuid = await get_task_manager(request.app).submit_task(
+        SEARCH_EXECUTION_METADATA,
         owner_metadata=OwnerMetadata.model_validate(
             WebServerOwnerMetadata(
                 user_id=_req_ctx.user_id,
@@ -577,14 +575,15 @@ class _PathParams(BaseModel):
         ),
         project_id=search_body.filters.project_id,
     )
-    _job_id = f"{async_job_rpc_get.job_id}"
+
+    _task_id = f"{task_uuid}"
     return create_data_response(
         TaskGet(
-            task_id=_job_id,
-            task_name=async_job_rpc_get.job_name,
-            status_href=f"{request.url.with_path(str(request.app.router['get_async_job_status'].url_for(task_id=_job_id)))}",
-            abort_href=f"{request.url.with_path(str(request.app.router['cancel_async_job'].url_for(task_id=_job_id)))}",
-            result_stream_href=f"{request.url.with_path(str(request.app.router['get_async_job_stream'].url_for(task_id=_job_id)))}",
+            task_id=_task_id,
+            task_name=SEARCH_TASK_NAME,
+            status_href=f"{request.url.with_path(str(request.app.router['get_async_job_status'].url_for(task_id=_task_id)))}",
+            abort_href=f"{request.url.with_path(str(request.app.router['cancel_async_job'].url_for(task_id=_task_id)))}",
+            result_stream_href=f"{request.url.with_path(str(request.app.router['get_async_job_stream'].url_for(task_id=_task_id)))}",
         ),
         status=status.HTTP_202_ACCEPTED,
     )