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