[DOP-21799] Refactor celery initialization

Author: Ilyas Gasanov

syncmaster/backend/__init__.py

@@ -32,7 +32,6 @@ def application_factory(settings: Settings) -> FastAPI:
         redoc_url=None,
     )
     application.state.settings = settings
-    application.state.celery = celery_factory(settings)
     application.include_router(api_router)
     application.exception_handler(RequestValidationError)(validation_exception_handler)
     application.exception_handler(ValidationError)(validation_exception_handler)
@@ -47,7 +46,6 @@ def application_factory(settings: Settings) -> FastAPI:
         {
             Settings: lambda: settings,
             UnitOfWork: get_uow(session_factory, settings=settings),
-            Celery: lambda: application.state.celery,
         },
     )
 

syncmaster/backend/api/v1/runs.py

@@ -5,11 +5,11 @@
 from typing import Annotated
 
 from asgi_correlation_id import correlation_id
-from celery import Celery
 from fastapi import APIRouter, Depends, Query
 from jinja2 import Template
 from kombu.exceptions import KombuError
 
+from syncmaster.backend.celery import app as celery
 from syncmaster.backend.dependencies import Stub
 from syncmaster.backend.services import UnitOfWork, get_user
 from syncmaster.backend.settings import ServerAppSettings as Settings
@@ -84,7 +84,6 @@ async def read_run(
 async def start_run(
     create_run_data: CreateRunSchema,
     settings: Annotated[Settings, Depends(Stub(Settings))],
-    celery: Annotated[Celery, Depends(Stub(Celery))],
     unit_of_work: UnitOfWork = Depends(UnitOfWork),
     current_user: User = Depends(get_user(is_active=True)),
 ) -> ReadRunSchema:

syncmaster/backend/celery.py

@@ -0,0 +1,6 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from syncmaster.backend.settings import ServerAppSettings
+from syncmaster.worker import celery_factory
+
+app = celery_factory(ServerAppSettings())

tests/test_unit/test_runs/test_create_run.py

@@ -20,7 +20,7 @@ async def test_developer_plus_can_create_run_of_transfer_his_group(
 ) -> None:
     # Arrange
     user = group_transfer.owner_group.get_member_of_role(role_developer_plus)
-    mock_send_task = mocker.patch("syncmaster.worker.celery.app.send_task")
+    mock_send_task = mocker.patch("syncmaster.backend.celery.app.send_task")
     mock_to_thread = mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     run = (
@@ -73,7 +73,7 @@ async def test_groupless_user_cannot_create_run(
     mocker,
 ) -> None:
     # Arrange
-    mocker.patch("syncmaster.worker.celery.app.send_task")
+    mocker.patch("syncmaster.backend.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -103,7 +103,7 @@ async def test_group_member_cannot_create_run_of_other_group_transfer(
     role_guest_plus: UserTestRoles,
 ):
     # Arrange
-    mocker.patch("syncmaster.worker.celery.app.send_task")
+    mocker.patch("syncmaster.backend.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
     user = group.get_member_of_role(role_guest_plus)
 
@@ -139,7 +139,7 @@ async def test_superuser_can_create_run(
     mocker,
 ) -> None:
     # Arrange
-    mock_send_task = mocker.patch("syncmaster.worker.celery.app.send_task")
+    mock_send_task = mocker.patch("syncmaster.backend.celery.app.send_task")
     mock_to_thread = mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -183,7 +183,7 @@ async def test_unauthorized_user_cannot_create_run(
     mocker,
 ) -> None:
     # Arrange
-    mocker.patch("syncmaster.worker.celery.app.send_task")
+    mocker.patch("syncmaster.backend.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -212,7 +212,7 @@ async def test_group_member_cannot_create_run_of_unknown_transfer_error(
 ) -> None:
     # Arrange
     user = group_transfer.owner_group.get_member_of_role(role_guest_plus)
-    mocker.patch("syncmaster.worker.celery.app.send_task")
+    mocker.patch("syncmaster.backend.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -240,7 +240,7 @@ async def test_superuser_cannot_create_run_of_unknown_transfer_error(
     mocker,
 ) -> None:
     # Arrange
-    mocker.patch("syncmaster.worker.celery.app.send_task")
+    mocker.patch("syncmaster.backend.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act

tests/test_unit/test_scheduler/test_transfer_job_manager.py

@@ -80,7 +80,7 @@ async def test_send_job_to_celery_with_success(
     group_transfer: MockTransfer,
 ):
     # Arrange
-    mock_send_task = mocker.patch("syncmaster.worker.celery.app.send_task")
+    mock_send_task = mocker.patch("syncmaster.scheduler.celery.app.send_task")
     mock_to_thread = mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -107,7 +107,7 @@ async def test_send_job_to_celery_with_failure(
     group_transfer: MockTransfer,
 ):
     # Arrange
-    mocker.patch("syncmaster.worker.celery.app.send_task")
+    mocker.patch("syncmaster.scheduler.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock, side_effect=KombuError)
 
     # Act & Assert