feat: 배치 잡 강제 실행 API 구현

waffledotcom/src/apps/batch/schemas.py

@@ -15,3 +15,14 @@ class JobDto(BaseModel):
 
 class ScheduleResponse(BaseModel):
     jobs: list[JobDto]
+
+
+class ForceRunResult(BaseModel):
+    name: str
+    tags: list[str]
+    success: bool
+    reason: str | None = None
+
+
+class ForceRunResponse(BaseModel):
+    results: list[ForceRunResult]

waffledotcom/src/apps/batch/views.py

@@ -1,8 +1,8 @@
 from fastapi import APIRouter
 
-from waffledotcom.src.batch.scheduler import scheduler
+from waffledotcom.src.batch.scheduler import get_job_name, scheduler
 
-from .schemas import JobDto, ScheduleResponse
+from .schemas import ForceRunResponse, ForceRunResult, JobDto, ScheduleResponse
 
 v1_router = APIRouter(prefix="/v1/batch", tags=["batch"])
 
@@ -24,3 +24,37 @@ def get_schedule(
         for job in scheduler.get_jobs(tag)
     ]
     return ScheduleResponse(jobs=job_dtos)
+
+
+@v1_router.post("/force-run", response_model_exclude_none=True)
+def force_run_job(
+    name: str | None = None,
+    tag: str | None = None,
+) -> ForceRunResponse:
+    if not ((name is None) ^ (tag is None)):
+        raise ValueError("Either name or tag should be provided")
+    jobs = scheduler.get_jobs(tag)
+    if name is not None:
+        jobs = [job for job in jobs if get_job_name(job) == name]
+
+    results: list[ForceRunResult] = []
+    for job in jobs:
+        try:
+            job.run()
+            results.append(
+                ForceRunResult(
+                    name=get_job_name(job),
+                    tags=list(map(str, job.tags)),
+                    success=True,
+                )
+            )
+        except Exception as e:
+            results.append(
+                ForceRunResult(
+                    name=get_job_name(job),
+                    tags=list(map(str, job.tags)),
+                    success=False,
+                    reason=str(e),
+                )
+            )
+    return ForceRunResponse(results=results)

waffledotcom/src/batch/scheduler.py

@@ -8,7 +8,7 @@
 from functools import wraps
 
 from loguru import logger
-from schedule import Scheduler
+from schedule import Job, Scheduler
 
 from waffledotcom.src.batch.slack.main import create_users_from_slack
 from waffledotcom.src.settings import settings
@@ -17,6 +17,10 @@
 scheduler = Scheduler()
 
 
+def get_job_name(job: Job) -> str:
+    return getattr(job.job_func, "__qualname__", "Unknown")
+
+
 def job_wrapper(job_func: Callable):
     job_name = job_func.__qualname__
 
@@ -49,7 +53,7 @@ def setup_job_schedule():
         ).tag("slack")
 
     for job in scheduler.get_jobs():
-        job_name = getattr(job.job_func, "__qualname__", "Unknown")
+        job_name = get_job_name(job)
 
         logger.info(
             "Job [{job_name}] is scheduled every {interval} {unit}. Next run at"