renamed

Author: Andrei Neagu

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

@@ -7,7 +7,7 @@
 
 from ...dsm import get_dsm_provider
 from ...modules.celery.models import TaskID
-from ...modules.celery.utils import get_celery_worker_client, get_fastapi_app
+from ...modules.celery.utils import get_celery_worker, get_fastapi_app
 from ...simcore_s3_dsm import SimcoreS3DataManager
 from ._progress_utils import get_tqdm_progress, set_tqdm_absolute_progress
 
@@ -32,9 +32,7 @@ async def data_export(
 
         async def _progress_cb(report: ProgressReport) -> None:
             set_tqdm_absolute_progress(pbar, report)
-            await get_celery_worker_client(task.app).set_task_progress(
-                task, task_id, report
-            )
+            await get_celery_worker(task.app).set_task_progress(task, task_id, report)
 
         return await dsm.create_s3_export(
             user_id, paths_to_export, progress_cb=_progress_cb

services/storage/src/simcore_service_storage/modules/celery/signals.py

@@ -13,7 +13,7 @@
 from ...modules.celery import get_event_loop, set_event_loop
 from ...modules.celery.utils import (
     get_fastapi_app,
-    set_celery_worker_client,
+    set_celery_worker,
     set_fastapi_app,
 )
 from ...modules.celery.worker import CeleryTaskQueueWorker
@@ -48,7 +48,7 @@ async def lifespan():
         set_event_loop(fastapi_app, loop)
 
         set_fastapi_app(sender.app, fastapi_app)
-        set_celery_worker_client(sender.app, CeleryTaskQueueWorker(sender.app))
+        set_celery_worker(sender.app, CeleryTaskQueueWorker(sender.app))
 
         loop.run_forever()
 

services/storage/src/simcore_service_storage/modules/celery/utils.py

@@ -3,15 +3,15 @@
 
 from .worker import CeleryTaskQueueWorker
 
-_WORKER_KEY = "celery_worker_client"
+_WORKER_KEY = "celery_worker"
 _FASTAPI_APP_KEY = "fastapi_app"
 
 
-def set_celery_worker_client(celery_app: Celery, worker: CeleryTaskQueueWorker) -> None:
+def set_celery_worker(celery_app: Celery, worker: CeleryTaskQueueWorker) -> None:
     celery_app.conf[_WORKER_KEY] = worker
 
 
-def get_celery_worker_client(celery_app: Celery) -> CeleryTaskQueueWorker:
+def get_celery_worker(celery_app: Celery) -> CeleryTaskQueueWorker:
     worker = celery_app.conf[_WORKER_KEY]
     assert isinstance(worker, CeleryTaskQueueWorker)
     return worker

services/storage/tests/unit/api/_worker_tasks/conftest.py

@@ -4,7 +4,7 @@
 from models_library.progress_bar import ProgressReport
 from pytest_mock import MockerFixture
 from simcore_service_storage.modules.celery.utils import (
-    set_celery_worker_client,
+    set_celery_worker,
     set_fastapi_app,
 )
 from simcore_service_storage.modules.celery.worker import CeleryTaskQueueWorker
@@ -15,7 +15,7 @@ def fake_celery_task(celery_app: Celery, initialized_app: FastAPI) -> Task:
     celery_task = Task()
     celery_task.app = celery_app
     set_fastapi_app(celery_app, initialized_app)
-    set_celery_worker_client(celery_app, CeleryTaskQueueWorker(celery_app))
+    set_celery_worker(celery_app, CeleryTaskQueueWorker(celery_app))
     return celery_task
 
 

services/storage/tests/unit/modules/celery/test_celery.py

@@ -20,7 +20,7 @@
     TaskState,
 )
 from simcore_service_storage.modules.celery.utils import (
-    get_celery_worker_client,
+    get_celery_worker,
     get_fastapi_app,
 )
 from tenacity import Retrying, retry_if_exception_type, stop_after_delay, wait_fixed
@@ -29,7 +29,7 @@
 
 
 async def _async_archive(celery_app: Celery, task: Task, files: list[str]) -> str:
-    worker_client = get_celery_worker_client(celery_app)
+    worker_client = get_celery_worker(celery_app)
 
     def sleep_for(seconds: float) -> None:
         time.sleep(seconds)