Switch from flake8 to ruff

Author: dolfinus

syncmaster/db/models/run.py

@@ -12,6 +12,7 @@
 
 from syncmaster.db.mixins import TimestampMixin
 from syncmaster.db.models.base import Base
+from syncmaster.db.models.group import Group
 from syncmaster.db.models.transfer import Transfer
 
 
@@ -65,7 +66,8 @@ class Run(Base, TimestampMixin):
     log_url: Mapped[str] = mapped_column(String(512), nullable=True)
     transfer_dump: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default={})
 
-    transfer: Mapped[Transfer] = relationship(Transfer)
+    transfer: Mapped[Transfer] = relationship(viewonly=True)
+    group: Mapped[Group] = relationship(viewonly=True)
 
     def __repr__(self):
         return (

syncmaster/db/models/transfer.py

@@ -20,7 +20,6 @@
 from syncmaster.db.mixins import ResourceMixin, TimestampMixin
 from syncmaster.db.models.base import Base
 from syncmaster.db.models.connection import Connection
-from syncmaster.db.models.queue import Queue
 
 
 class Transfer(
@@ -60,7 +59,6 @@ class Transfer(
 
     source_connection: Mapped[Connection] = relationship(foreign_keys=source_connection_id)
     target_connection: Mapped[Connection] = relationship(foreign_keys=target_connection_id)
-    queue: Mapped[Queue] = relationship()
 
     search_vector: Mapped[str] = mapped_column(
         TSVECTOR,

syncmaster/scheduler/transfer_job_manager.py

@@ -89,6 +89,7 @@ async def send_job_to_celery(transfer_id: int) -> None:
             except TransferNotFoundError:
                 return
 
+            queue = await unit_of_work.queue.read_by_id(transfer.queue_id)
             credentials_source = await unit_of_work.credentials.read(transfer.source_connection_id)
             credentials_target = await unit_of_work.credentials.read(transfer.target_connection_id)
 
@@ -105,7 +106,7 @@ async def send_job_to_celery(transfer_id: int) -> None:
                     celery.send_task,
                     "run_transfer_task",
                     kwargs={"run_id": run.id},
-                    queue=transfer.queue.slug,
+                    queue=queue.slug,
                 )
             except KombuError as e:
                 async with unit_of_work:

syncmaster/server/api/v1/runs.py

@@ -98,6 +98,7 @@ async def start_run(
         raise ActionNotAllowedError
 
     transfer = await unit_of_work.transfer.read_by_id(transfer_id=create_run_data.transfer_id)
+    queue = await unit_of_work.queue.read_by_id(transfer.queue_id)
 
     # The credentials.read method is used rather than credentials.read_bulk deliberately
     # it's more convenient to transfer credits in this place
@@ -123,7 +124,7 @@ async def start_run(
             celery.send_task,
             "run_transfer_task",
             kwargs={"run_id": run.id},
-            queue=transfer.queue.slug,
+            queue=queue.slug,
         )
     except KombuError as e:
         async with unit_of_work:

syncmaster/worker/controller.py

@@ -194,7 +194,7 @@ def __init__(  # noqa: PLR0913
             run_data={"id": run.id, "created_at": run.created_at},
             transfer_id=run.transfer.id,
             transfer_name=run.transfer.name,
-            group_name=run.transfer.group.name,
+            group_name=run.group.name,
             transfer_params=run.transfer.source_params,
             strategy_params=run.transfer.strategy_params,
             resources=run.transfer.resources,
@@ -207,7 +207,7 @@ def __init__(  # noqa: PLR0913
             run_data={"id": run.id, "created_at": run.created_at},
             transfer_id=run.transfer.id,
             transfer_name=run.transfer.name,
-            group_name=run.transfer.group.name,
+            group_name=run.group.name,
             transfer_params=run.transfer.target_params,
             strategy_params=run.transfer.strategy_params,
             resources=run.transfer.resources,

syncmaster/worker/spark.py

@@ -34,7 +34,7 @@ def get_worker_spark_session(
     """Construct Spark Session using run parameters and application settings"""
     from pyspark.sql import SparkSession  # noqa: PLC0415
 
-    name = run.transfer.group.name + "_" + run.transfer.name
+    name = run.group.name + "_" + run.transfer.name
     spark_builder = SparkSession.builder.appName(f"SyncMaster__{name}")
 
     master = settings.spark_session_default_config.get("spark.master")

syncmaster/worker/transfer.py

@@ -41,8 +41,8 @@ def run_transfer(run_id: int, engine: Engine, settings: WorkerAppSettings):
                     Run.status == Status.CREATED,
                 )
                 .options(
+                    selectinload(Run.group),
                     selectinload(Run.transfer),
-                    selectinload(Run.transfer).selectinload(Transfer.group),
                     selectinload(Run.transfer).selectinload(Transfer.source_connection),
                     selectinload(Run.transfer).selectinload(Transfer.target_connection),
                 ),

tests/test_unit/test_runs/test_create_run.py

@@ -7,7 +7,7 @@
 from sqlalchemy import desc, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from syncmaster.db.models import Run, RunType, Status
+from syncmaster.db.models import Queue, Run, RunType, Status
 from tests.mocks import MockGroup, MockTransfer, MockUser, UserTestRoles
 
 pytestmark = [pytest.mark.asyncio, pytest.mark.server]
@@ -46,6 +46,10 @@ async def test_developer_plus_can_create_run_of_transfer_his_group(
         )
     ).first()
 
+    queue = await session.scalar(
+        select(Queue).where(Queue.id == group_transfer.queue_id),
+    )
+
     assert response.json() == {
         "id": run.id,
         "transfer_id": run.transfer_id,
@@ -60,7 +64,7 @@ async def test_developer_plus_can_create_run_of_transfer_his_group(
         mock_send_task,
         "run_transfer_task",
         kwargs={"run_id": run.id},
-        queue=group_transfer.queue.slug,
+        queue=queue.slug,
     )
 
 
@@ -145,6 +149,10 @@ async def test_superuser_can_create_run(
         )
     ).first()
 
+    queue = await session.scalar(
+        select(Queue).where(Queue.id == group_transfer.queue_id),
+    )
+
     assert response.json() == {
         "id": run.id,
         "transfer_id": run.transfer_id,
@@ -159,7 +167,7 @@ async def test_superuser_can_create_run(
         mock_send_task,
         "run_transfer_task",
         kwargs={"run_id": run.id},
-        queue=group_transfer.queue.slug,
+        queue=queue.slug,
     )
 
 

tests/test_unit/test_scheduler/test_transfer_job_manager.py

@@ -7,7 +7,7 @@
 from sqlalchemy import desc, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from syncmaster.db.models import Run, Status
+from syncmaster.db.models import Queue, Run, Status
 from syncmaster.exceptions.run import CannotConnectToTaskQueueError
 from syncmaster.scheduler.transfer_job_manager import TransferJobManager
 from tests.mocks import MockTransfer
@@ -79,11 +79,15 @@ async def test_send_job_to_celery_with_success(
         )
     ).first()
 
+    queue = await session.scalar(
+        select(Queue).where(Queue.id == group_transfer.queue_id),
+    )
+
     mock_to_thread.assert_awaited_once_with(
         mock_send_task,
         "run_transfer_task",
         kwargs={"run_id": run.id},
-        queue=group_transfer.queue.slug,
+        queue=queue.slug,
     )