Spaces:
Running
Running
File size: 11,880 Bytes
947ea10 b7dddbe 947ea10 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 | from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query
from app.api.deps import get_scheduler_service
from app.core.logger import get_logger
from app.services.scheduler_service import (
SchedulerService,
job_to_response,
history_to_response,
validate_cron_expression,
validate_timezone,
validate_url,
)
router = APIRouter()
logger = get_logger(__name__)
def _paginate(items: list[Any], total: int, page: int, page_size: int) -> dict[str, Any]:
return {
"items": items,
"total": total,
"page": page,
"page_size": page_size,
"has_next": (page * page_size) < total,
"has_prev": page > 1,
}
# ---------------------------------------------------------------------------
# Scheduler Monitoring
# ---------------------------------------------------------------------------
@router.get("/scheduler/status", summary="Get scheduler status")
async def get_scheduler_status(
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
status = scheduler_service.get_scheduler_status()
status["running_job_ids"] = scheduler_service.get_running_job_ids()
return {"success": True, "data": status}
@router.get("/scheduler/metrics", summary="Get scheduler metrics")
async def get_scheduler_metrics(
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
metrics = await scheduler_service.get_metrics()
return {"success": True, "data": metrics}
@router.get("/scheduler/health", summary="Get scheduler health (Redis + instance status)")
async def get_scheduler_health(
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
health = await scheduler_service.get_health()
return {"success": True, "data": health}
# ---------------------------------------------------------------------------
# Job CRUD
# ---------------------------------------------------------------------------
@router.post("/scheduler/jobs", summary="Create a new scheduled job", status_code=201)
async def create_job(
body: dict[str, Any],
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
errors = _validate_job_create(body)
if errors:
raise HTTPException(status_code=422, detail={"success": False, "errors": errors})
try:
job = await scheduler_service.create_job(body)
return {"success": True, "data": job_to_response(job)}
except ValueError as exc:
raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
@router.get("/scheduler/jobs", summary="List all scheduled jobs")
async def list_jobs(
status: str | None = Query(None, description="Filter by status (active, paused, completed, failed, deleted)"),
tags: str | None = Query(None, description="Comma-separated tags to filter by"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(20, ge=1, le=100, description="Items per page"),
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
tag_list = [t.strip() for t in tags.split(",") if t.strip()] if tags else None
jobs, total = await scheduler_service.list_jobs(
status=status, tags=tag_list, page=page, page_size=page_size,
)
return {
"success": True,
"data": _paginate(
items=[job_to_response(j) for j in jobs],
total=total, page=page, page_size=page_size,
),
}
@router.get("/scheduler/jobs/{job_id}", summary="Get a scheduled job by ID")
async def get_job(
job_id: str,
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
job = await scheduler_service.get_job(job_id)
if not job or job.get("status") == "deleted":
raise HTTPException(status_code=404, detail={"success": False, "error": "Job not found"})
return {"success": True, "data": job_to_response(job)}
@router.put("/scheduler/jobs/{job_id}", summary="Update a scheduled job")
async def update_job(
job_id: str,
body: dict[str, Any],
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
try:
job = await scheduler_service.update_job(job_id, body)
return {"success": True, "data": job_to_response(job)}
except KeyError as exc:
raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
@router.delete("/scheduler/jobs/{job_id}", summary="Delete a scheduled job (soft delete)")
async def delete_job(
job_id: str,
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
try:
await scheduler_service.delete_job(job_id)
return {"success": True, "message": "Job deleted successfully"}
except KeyError as exc:
raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
@router.delete("/scheduler/jobs/{job_id}/hard", summary="Permanently delete a scheduled job")
async def hard_delete_job(
job_id: str,
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
try:
await scheduler_service.hard_delete_job(job_id)
return {"success": True, "message": "Job permanently deleted"}
except KeyError as exc:
raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
# ---------------------------------------------------------------------------
# Job Control
# ---------------------------------------------------------------------------
@router.post("/scheduler/jobs/{job_id}/pause", summary="Pause a scheduled job")
async def pause_job(
job_id: str,
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
try:
job = await scheduler_service.pause_job(job_id)
return {"success": True, "data": job_to_response(job)}
except KeyError as exc:
raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
@router.post("/scheduler/jobs/{job_id}/resume", summary="Resume a paused job")
async def resume_job(
job_id: str,
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
try:
job = await scheduler_service.resume_job(job_id)
return {"success": True, "data": job_to_response(job)}
except KeyError as exc:
raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
@router.post("/scheduler/jobs/{job_id}/run", summary="Trigger a job execution immediately")
async def run_job_now(
job_id: str,
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
try:
await scheduler_service.run_job_now(job_id)
return {"success": True, "message": "Job execution triggered"}
except KeyError as exc:
raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
# ---------------------------------------------------------------------------
# Execution History
# ---------------------------------------------------------------------------
@router.get("/scheduler/jobs/{job_id}/history", summary="Get execution history for a job")
async def get_job_history(
job_id: str,
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(20, ge=1, le=100, description="Items per page"),
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
job = await scheduler_service.get_job(job_id)
if not job:
raise HTTPException(status_code=404, detail={"success": False, "error": "Job not found"})
history, total = await scheduler_service.get_job_history(
job_id=job_id, page=page, page_size=page_size,
)
return {
"success": True,
"data": _paginate(
items=[history_to_response(h) for h in history],
total=total, page=page, page_size=page_size,
),
}
@router.get("/scheduler/history", summary="Get global execution history")
async def get_execution_history(
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(50, ge=1, le=200, description="Items per page"),
status: str | None = Query(None, description="Filter by execution status"),
scheduler_service: SchedulerService = Depends(get_scheduler_service),
):
history, total = await scheduler_service.get_execution_history(
page=page, page_size=page_size, status=status,
)
return {
"success": True,
"data": _paginate(
items=[history_to_response(h) for h in history],
total=total, page=page, page_size=page_size,
),
}
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
def _validate_job_create(body: dict[str, Any]) -> list[str]:
errors: list[str] = []
if not body.get("name"):
errors.append("name is required")
elif not isinstance(body["name"], str) or len(body["name"]) > 255:
errors.append("name must be a string between 1 and 255 characters")
if not body.get("url"):
errors.append("url is required")
elif not isinstance(body["url"], str):
errors.append("url must be a string")
else:
try:
validate_url(body["url"])
except ValueError as exc:
errors.append(str(exc))
trigger = body.get("trigger")
if not trigger:
errors.append("trigger is required")
elif not isinstance(trigger, dict):
errors.append("trigger must be an object")
else:
trigger_type = trigger.get("type")
if not trigger_type:
errors.append("trigger.type is required")
elif trigger_type not in ("cron", "interval", "date"):
errors.append("trigger.type must be one of: cron, interval, date")
elif trigger_type == "cron":
if not trigger.get("cron_expression"):
errors.append("trigger.cron_expression is required for cron trigger")
else:
try:
validate_cron_expression(trigger["cron_expression"])
except ValueError as exc:
errors.append(str(exc))
if body.get("timezone"):
try:
validate_timezone(body["timezone"])
except ValueError as exc:
errors.append(str(exc))
timeout = body.get("timeout")
if timeout is not None:
if not isinstance(timeout, (int, float)):
errors.append("timeout must be a number")
elif timeout < 1 or timeout > 300:
errors.append("timeout must be between 1 and 300")
method = body.get("method")
if method and method not in ("GET", "POST", "PUT", "PATCH", "DELETE"):
errors.append("method must be one of: GET, POST, PUT, PATCH, DELETE")
retry = body.get("retry")
if retry is not None:
if not isinstance(retry, dict):
errors.append("retry must be an object")
else:
max_retries = retry.get("max_retries")
if max_retries is not None and (not isinstance(max_retries, int) or max_retries < 0 or max_retries > 10):
errors.append("retry.max_retries must be between 0 and 10")
return errors
|