extract AA CLI arg to constants

app/shared/constants.py

@@ -2,3 +2,6 @@
 STATUS_FAILURE = "FAILURE"
 STATUS_PENDING = "PENDING"
 STATUS_SUCCESS = "SUCCESS"
+
+# AA CLI CONFIGS
+SHEET_ID = "--gsheet_feeder_db.sheet_id"

app/tests/worker/test_worker_main.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 import pytest
 from auto_archiver.core import Media, Metadata
 
-from app.shared import schemas
+from app.shared import constants, schemas
 from app.shared.db import models
 from app.web.utils.misc import get_all_urls
 from app.worker.main import create_archive_task, create_sheet_task
@@ -119,7 +119,7 @@ class TestCreateSheetTask:
         res = create_sheet_task(self.sheet.model_dump_json())
 
         m_args.assert_called_once_with(
-            "interstellar", True, ["--gsheet_feeder_db.sheet_id", "123"]
+            "interstellar", True, [constants.SHEET_ID, "123"]
         )
         m_orchestrator.return_value.setup.assert_called_once()
         m_orchestrator.return_value.feed.assert_called_once()

app/worker/main.py

@@ -7,7 +7,7 @@ from celery.signals import task_failure
 from loguru import logger
 from sqlalchemy import exc
 
-from app.shared import business_logic, schemas
+from app.shared import business_logic, constants, schemas
 from app.shared.db import models, worker_crud
 from app.shared.db.database import get_db
 from app.shared.log import log_error
@@ -78,7 +78,7 @@ def create_sheet_task(self, sheet_json: str):
     logger.info(f"[queue={queue_name}] SHEET START {sheet=}")
 
     args = get_orchestrator_args(
-        sheet.group_id, True, ["--gsheet_feeder_db.sheet_id", sheet.sheet_id]
+        sheet.group_id, True, [constants.SHEET_ID, sheet.sheet_id]
     )
     orchestrator = ArchivingOrchestrator()
     orchestrator.logger_id = AA_LOGGER_ID  # ensure single logger