From c2d56d74754703e2f6cff71161f0bfd33eec20a8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 9 Mar 2026 18:01:34 +0000 Subject: [PATCH] feat(tasks): add scheduled batch processing infrastructure (model, migration, tasks, API, view, template) Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/scheduled_jobs.py | 258 ++++++++ app/celery_worker.py | 5 + app/models.py | 56 ++ app/tasks/batch_tasks.py | 325 ++++++++++ app/views/scheduled_jobs.py | 44 ++ frontend/templates/admin_scheduled_jobs.html | 558 ++++++++++++++++++ migrations/versions/026_add_scheduled_jobs.py | 51 ++ 7 files changed, 1297 insertions(+) create mode 100644 app/api/scheduled_jobs.py create mode 100644 app/tasks/batch_tasks.py create mode 100644 app/views/scheduled_jobs.py create mode 100644 frontend/templates/admin_scheduled_jobs.html create mode 100644 migrations/versions/026_add_scheduled_jobs.py diff --git a/app/api/scheduled_jobs.py b/app/api/scheduled_jobs.py new file mode 100644 index 00000000..7292d8d1 --- /dev/null +++ b/app/api/scheduled_jobs.py @@ -0,0 +1,258 @@ +""" +Admin API endpoints for managing scheduled batch processing jobs. + +All endpoints require admin privileges (checked via session ``is_admin`` flag). + +Available routes: + GET /api/admin/scheduled-jobs – list all scheduled jobs + PATCH /api/admin/scheduled-jobs/{id} – update schedule / enable-disable + POST /api/admin/scheduled-jobs/{id}/run-now – trigger a job immediately +""" + +import logging +from datetime import datetime, timezone +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from app.database import get_db +from app.models import ScheduledJob + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/admin/scheduled-jobs", tags=["admin-scheduled-jobs"]) + +DbSession = Annotated[Session, Depends(get_db)] + +# --------------------------------------------------------------------------- +# Authorisation helper +# --------------------------------------------------------------------------- + + +def _require_admin(request: Request) -> dict: + """Ensure the caller is an admin; raises HTTP 403 otherwise.""" + user = request.session.get("user") + if not user or not user.get("is_admin"): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") + return user + + +AdminUser = Annotated[dict, Depends(_require_admin)] + + +# --------------------------------------------------------------------------- +# Pydantic schemas +# --------------------------------------------------------------------------- + + +class ScheduledJobResponse(BaseModel): + """Read model for a scheduled job.""" + + id: int + name: str + display_name: str + description: str | None + task_name: str + enabled: bool + schedule_type: str + cron_minute: str + cron_hour: str + cron_day_of_week: str + cron_day_of_month: str + cron_month_of_year: str + interval_seconds: int | None + last_run_at: datetime | None + last_run_status: str | None + last_run_detail: str | None + created_at: datetime | None + updated_at: datetime | None + + class Config: + from_attributes = True + + +class ScheduledJobUpdate(BaseModel): + """Writable fields for a scheduled job update (all optional).""" + + enabled: bool | None = Field(None, description="Whether the job is active") + schedule_type: str | None = Field(None, pattern="^(cron|interval)$", description="'cron' or 'interval'") + cron_minute: str | None = Field(None, max_length=50) + cron_hour: str | None = Field(None, max_length=50) + cron_day_of_week: str | None = Field(None, max_length=50) + cron_day_of_month: str | None = Field(None, max_length=50) + cron_month_of_year: str | None = Field(None, max_length=50) + interval_seconds: int | None = Field(None, ge=60, description="Interval in seconds (min 60)") + + +# --------------------------------------------------------------------------- +# Default job definitions – seeded into the DB on first startup +# --------------------------------------------------------------------------- + +DEFAULT_JOBS: list[dict[str, Any]] = [ + { + "name": "process-new-documents", + "display_name": "Process New Documents", + "description": ( + "Scans for documents that have been uploaded but never processed " + "and queues them through the full processing pipeline. " + "Runs hourly by default." + ), + "task_name": "app.tasks.batch_tasks.process_new_documents", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "0", + "cron_hour": "*/1", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "reprocess-failed-documents", + "display_name": "Reprocess Failed Documents", + "description": ( + "Finds documents whose last processing attempt failed and re-queues " + "them for reprocessing. Only picks up files that are not currently " + "being processed. Runs every 6 hours by default." + ), + "task_name": "app.tasks.batch_tasks.reprocess_failed_documents", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "30", + "cron_hour": "*/6", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "cleanup-temp-files", + "display_name": "Clean Up Temporary Files", + "description": ( + "Removes stale files from the workdir/tmp directory. " + "Only files older than 24 hours that are not referenced by any active " + "processing job are deleted. Runs daily at 03:30 UTC by default." + ), + "task_name": "app.tasks.batch_tasks.cleanup_temp_files", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "30", + "cron_hour": "3", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, +] + + +def seed_default_scheduled_jobs(db: Session) -> None: + """ + Insert the built-in scheduled jobs if they do not already exist. + + Called from the FastAPI lifespan handler so the records are available + immediately after the first startup. + """ + for job_data in DEFAULT_JOBS: + existing = db.query(ScheduledJob).filter(ScheduledJob.name == job_data["name"]).first() + if existing is None: + db.add(ScheduledJob(**job_data)) + try: + db.commit() + except Exception as exc: + db.rollback() + logger.error("Failed to seed default scheduled jobs: %s", exc) + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.get("", response_model=list[ScheduledJobResponse]) +def list_scheduled_jobs(request: Request, db: DbSession, _admin: AdminUser) -> list[ScheduledJobResponse]: + """ + Return all scheduled jobs ordered by display name. + + Requires admin privileges. + """ + jobs = db.query(ScheduledJob).order_by(ScheduledJob.display_name).all() + return jobs # type: ignore[return-value] + + +@router.patch("/{job_id}", response_model=ScheduledJobResponse) +def update_scheduled_job( + job_id: int, + payload: ScheduledJobUpdate, + request: Request, + db: DbSession, + _admin: AdminUser, +) -> ScheduledJobResponse: + """ + Update schedule configuration or enabled state for a job. + + Only the fields included in the request body are modified. + Changes to the Celery Beat schedule take effect after the worker restarts. + + Requires admin privileges. + """ + job = db.query(ScheduledJob).filter(ScheduledJob.id == job_id).first() + if job is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled job not found") + + update_data = payload.model_dump(exclude_none=True) + if not update_data: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No fields to update") + + for field, value in update_data.items(): + setattr(job, field, value) + + job.updated_at = datetime.now(timezone.utc) + + try: + db.commit() + db.refresh(job) + except Exception as exc: + db.rollback() + logger.error("Failed to update scheduled job %s: %s", job_id, exc) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to update scheduled job", + ) from exc + + logger.info("Admin updated scheduled job %s (id=%s): %s", job.name, job_id, update_data) + return job # type: ignore[return-value] + + +@router.post("/{job_id}/run-now") +def run_scheduled_job_now( + job_id: int, + request: Request, + db: DbSession, + _admin: AdminUser, +) -> dict[str, Any]: + """ + Immediately dispatch the Celery task for the given scheduled job. + + The task is sent to the default queue; its result is tracked asynchronously + via the ``last_run_at`` / ``last_run_status`` fields updated by the task + itself. + + Requires admin privileges. + """ + job = db.query(ScheduledJob).filter(ScheduledJob.id == job_id).first() + if job is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled job not found") + + from app.celery_app import celery as celery_app + + task = celery_app.send_task(job.task_name) + logger.info("Admin triggered scheduled job %s (id=%s) manually, task_id=%s", job.name, job_id, task.id) + + return { + "status": "dispatched", + "job_id": job_id, + "job_name": job.name, + "task_id": task.id, + } diff --git a/app/celery_worker.py b/app/celery_worker.py index 378b71af..ca9d11f0 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -9,6 +9,11 @@ from app import tasks # noqa: F401 - Imports app/tasks.py so Celery can registe from app.celery_app import celery from app.config import settings from app.tasks.backup_tasks import cleanup_old_backups, create_backup # noqa: F401 +from app.tasks.batch_tasks import ( # noqa: F401 + cleanup_temp_files, + process_new_documents, + reprocess_failed_documents, +) from app.tasks.check_credentials import check_credentials from app.tasks.compute_embedding import backfill_missing_embeddings, compute_document_embedding # noqa: F401 from app.tasks.convert_to_pdf import convert_to_pdf # noqa: F401 diff --git a/app/models.py b/app/models.py index 4e5bd6ef..0cc7b53a 100644 --- a/app/models.py +++ b/app/models.py @@ -777,3 +777,59 @@ class InAppNotification(Base): is_read = Column(Boolean, nullable=False, default=False, index=True) file_id = Column(Integer, nullable=True) # Optional link to FileRecord created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True) + + +class ScheduledJob(Base): + """ + Configuration record for an admin-managed scheduled batch processing job. + + Each row represents one recurring job entry. The Celery Beat schedule is + built from these rows at worker startup; changes take effect after the + worker process is restarted. + + Schedule types + -------------- + - ``"cron"`` – standard cron expression fields (minute/hour/…) + - ``"interval"`` – fixed interval in seconds (e.g. 3600 for hourly) + """ + + __tablename__ = "scheduled_jobs" + + id = Column(Integer, primary_key=True, index=True) + + # Unique machine-readable key used as the Celery Beat schedule entry name. + name = Column(String(100), unique=True, nullable=False, index=True) + + # Human-readable display name shown in the admin UI. + display_name = Column(String(255), nullable=False) + + # Short description of what the job does. + description = Column(Text, nullable=True) + + # Fully-qualified Celery task name, e.g. "app.tasks.batch_tasks.process_new_documents". + task_name = Column(String(255), nullable=False) + + # Whether the job is active. Inactive jobs are excluded from the beat schedule. + enabled = Column(Boolean, nullable=False, default=True) + + # Schedule type: "cron" or "interval". + schedule_type = Column(String(20), nullable=False, default="cron") + + # --- Cron fields (used when schedule_type == "cron") --- + cron_minute = Column(String(50), nullable=False, default="0") + cron_hour = Column(String(50), nullable=False, default="*") + cron_day_of_week = Column(String(50), nullable=False, default="*") + cron_day_of_month = Column(String(50), nullable=False, default="*") + cron_month_of_year = Column(String(50), nullable=False, default="*") + + # --- Interval field (used when schedule_type == "interval") --- + # Interval in seconds; e.g. 3600 = hourly, 86400 = daily. + interval_seconds = Column(Integer, nullable=True) + + # Timestamps populated by the worker after each execution. + last_run_at = Column(DateTime(timezone=True), nullable=True) + last_run_status = Column(String(20), nullable=True) # "success", "failed", "running" + last_run_detail = Column(Text, nullable=True) # Brief result summary or error + + created_at = Column(DateTime(timezone=True), server_default=func.now()) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/app/tasks/batch_tasks.py b/app/tasks/batch_tasks.py new file mode 100644 index 00000000..ae18d2c8 --- /dev/null +++ b/app/tasks/batch_tasks.py @@ -0,0 +1,325 @@ +""" +Scheduled batch processing tasks for DocuElevate. + +This module provides three Celery tasks that can be scheduled via Celery Beat +and managed through the admin UI (``/admin/scheduled-jobs``): + +- ``process_new_documents`` – Queue any documents that have never been processed. +- ``reprocess_failed_documents`` – Re-queue documents whose processing failed. +- ``cleanup_temp_files`` – Remove stale files from the ``workdir/tmp`` directory. + +Each task records its execution result back to the ``ScheduledJob`` table so +the admin UI can display last-run times and statuses. +""" + +import logging +import os +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from app.celery_app import celery +from app.config import settings +from app.database import SessionLocal +from app.models import FileProcessingStep, FileRecord, ScheduledJob + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +_PROCESSING_STEPS = { + "create_file_record", + "check_text", + "extract_text", + "process_with_ocr", + "process_with_azure_document_intelligence", + "extract_metadata_with_gpt", + "embed_metadata_into_pdf", + "finalize_document_storage", + "send_to_all_destinations", +} + + +def _update_job_status(job_name: str, status: str, detail: str) -> None: + """Persist run status back to the ScheduledJob row for display in the UI.""" + try: + with SessionLocal() as db: + job = db.query(ScheduledJob).filter(ScheduledJob.name == job_name).first() + if job: + job.last_run_at = datetime.now(timezone.utc) + job.last_run_status = status + job.last_run_detail = detail + db.commit() + except Exception as exc: # pragma: no cover – best-effort status update + logger.warning("Could not update ScheduledJob status for %s: %s", job_name, exc) + + +# --------------------------------------------------------------------------- +# Task: process new (unprocessed) documents +# --------------------------------------------------------------------------- + + +@celery.task(name="app.tasks.batch_tasks.process_new_documents") +def process_new_documents() -> dict: + """ + Queue all documents that have never been processed. + + A document is considered *new* when it has no ``FileProcessingStep`` rows + that match the core pipeline steps. The task loads each qualifying + ``FileRecord``, verifies that the original file still exists on disk, and + dispatches ``process_document`` for each one. + + Returns a summary dict with ``queued`` and ``skipped`` counts. + """ + from app.tasks.process_document import process_document # avoid circular import + + job_name = "process-new-documents" + logger.info("[batch] Starting process_new_documents task") + + try: + with SessionLocal() as db: + # Files that already have at least one processing step recorded. + processed_file_ids = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.step_name.in_(_PROCESSING_STEPS)) + .distinct() + .subquery() + ) + + # Candidate files: non-duplicate records with no processing steps yet. + candidates = ( + db.query(FileRecord) + .filter(FileRecord.is_duplicate.is_(False)) + .filter(~FileRecord.id.in_(db.query(processed_file_ids.c.file_id))) + .all() + ) + + queued = 0 + skipped = 0 + for record in candidates: + if not record.local_filename or not os.path.exists(record.local_filename): + logger.warning( + "[batch] Skipping file_id=%s — local file not found: %s", + record.id, + record.local_filename, + ) + skipped += 1 + continue + process_document.delay( + record.local_filename, + original_filename=record.original_filename, + file_id=record.id, + owner_id=record.owner_id, + ) + queued += 1 + + detail = f"Queued {queued} document(s) for processing; skipped {skipped} (file not on disk)." + logger.info("[batch] process_new_documents: %s", detail) + _update_job_status(job_name, "success", detail) + return {"queued": queued, "skipped": skipped} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] process_new_documents failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"queued": 0, "skipped": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: reprocess failed documents +# --------------------------------------------------------------------------- + + +@celery.task(name="app.tasks.batch_tasks.reprocess_failed_documents") +def reprocess_failed_documents() -> dict: + """ + Re-queue documents whose most-recent processing attempt failed. + + Only files that have at least one ``FileProcessingStep`` with + ``status == "failure"`` **and** no currently ``in_progress`` steps are + selected so that actively-running jobs are not interrupted. + + Returns a summary dict with ``queued`` and ``skipped`` counts. + """ + from app.tasks.process_document import process_document # avoid circular import + + job_name = "reprocess-failed-documents" + logger.info("[batch] Starting reprocess_failed_documents task") + + try: + with SessionLocal() as db: + # Files with at least one failed step. + failed_file_ids = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.step_name.in_(_PROCESSING_STEPS)) + .filter(FileProcessingStep.status == "failure") + .distinct() + .subquery() + ) + + # Exclude files that are currently being processed. + in_progress_file_ids = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.status == "in_progress") + .distinct() + .subquery() + ) + + candidates = ( + db.query(FileRecord) + .filter(FileRecord.is_duplicate.is_(False)) + .filter(FileRecord.id.in_(db.query(failed_file_ids.c.file_id))) + .filter(~FileRecord.id.in_(db.query(in_progress_file_ids.c.file_id))) + .all() + ) + + queued = 0 + skipped = 0 + for record in candidates: + if not record.local_filename or not os.path.exists(record.local_filename): + logger.warning( + "[batch] Skipping file_id=%s — local file not found: %s", + record.id, + record.local_filename, + ) + skipped += 1 + continue + process_document.delay( + record.local_filename, + original_filename=record.original_filename, + file_id=record.id, + owner_id=record.owner_id, + ) + queued += 1 + + detail = f"Re-queued {queued} failed document(s); skipped {skipped} (file not on disk)." + logger.info("[batch] reprocess_failed_documents: %s", detail) + _update_job_status(job_name, "success", detail) + return {"queued": queued, "skipped": skipped} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] reprocess_failed_documents failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"queued": 0, "skipped": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: clean up temporary files +# --------------------------------------------------------------------------- + +#: Files in ``workdir/tmp`` that are older than this threshold are deleted. +_TEMP_FILE_MAX_AGE_HOURS: int = 24 + + +@celery.task(name="app.tasks.batch_tasks.cleanup_temp_files") +def cleanup_temp_files(max_age_hours: int = _TEMP_FILE_MAX_AGE_HOURS) -> dict: + """ + Delete stale files from the ``workdir/tmp`` directory. + + A file is considered stale when **both** of the following are true: + + 1. Its modification time is older than *max_age_hours* (default 24 h). + 2. No ``FileRecord.local_filename`` points to it **or** the file is not + referenced by any active in-progress processing step. + + This prevents accidental deletion of files that are being actively + processed by the pipeline. + + Args: + max_age_hours: Minimum age (in hours) before a temp file is eligible + for deletion. Defaults to 24. + + Returns: + A summary dict with ``deleted`` and ``skipped`` counts. + """ + job_name = "cleanup-temp-files" + logger.info("[batch] Starting cleanup_temp_files (max_age_hours=%s)", max_age_hours) + + tmp_dir = Path(settings.workdir) / "tmp" + if not tmp_dir.exists(): + detail = "workdir/tmp does not exist; nothing to clean." + logger.info("[batch] cleanup_temp_files: %s", detail) + _update_job_status(job_name, "success", detail) + return {"deleted": 0, "skipped": 0} + + cutoff = datetime.now(timezone.utc) - timedelta(hours=max_age_hours) + + deleted = 0 + skipped = 0 + errors = 0 + + try: + with SessionLocal() as db: + # Collect filenames actively referenced by in-progress processing steps. + in_progress_filenames: set[str] = set() + in_progress_records = ( + db.query(FileRecord.local_filename) + .join(FileProcessingStep, FileProcessingStep.file_id == FileRecord.id) + .filter(FileProcessingStep.status == "in_progress") + .distinct() + .all() + ) + for row in in_progress_records: + if row.local_filename: + in_progress_filenames.add(os.path.basename(row.local_filename)) + + # Also collect all filenames referenced by FileRecord.local_filename + # that point into workdir/tmp (files still in the tmp pipeline). + active_tmp_filenames: set[str] = set() + tmp_dir_str = str(tmp_dir.resolve()) + active_records = ( + db.query(FileRecord.local_filename) + .filter(FileRecord.local_filename.like(f"{tmp_dir_str}%")) + .all() + ) + for row in active_records: + if row.local_filename: + active_tmp_filenames.add(os.path.basename(row.local_filename)) + + protected_basenames = in_progress_filenames | active_tmp_filenames + + for entry in tmp_dir.iterdir(): + if not entry.is_file(): + continue + + # Check modification time. + try: + mtime = datetime.fromtimestamp(entry.stat().st_mtime, tz=timezone.utc) + except OSError: + skipped += 1 + continue + + if mtime >= cutoff: + skipped += 1 + continue + + if entry.name in protected_basenames: + logger.debug("[batch] cleanup_temp_files: keeping protected file %s", entry.name) + skipped += 1 + continue + + try: + entry.unlink() + logger.debug("[batch] cleanup_temp_files: deleted %s", entry) + deleted += 1 + except OSError as exc: + logger.warning("[batch] cleanup_temp_files: could not delete %s: %s", entry, exc) + errors += 1 + + detail = ( + f"Deleted {deleted} stale temp file(s); " + f"skipped {skipped} (too new or protected); " + f"{errors} error(s)." + ) + status = "failed" if errors and not deleted else "success" + logger.info("[batch] cleanup_temp_files: %s", detail) + _update_job_status(job_name, status, detail) + return {"deleted": deleted, "skipped": skipped, "errors": errors} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] cleanup_temp_files failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"deleted": 0, "skipped": 0, "errors": 1, "error": str(exc)} diff --git a/app/views/scheduled_jobs.py b/app/views/scheduled_jobs.py new file mode 100644 index 00000000..a0d68bf8 --- /dev/null +++ b/app/views/scheduled_jobs.py @@ -0,0 +1,44 @@ +"""Admin view: scheduled batch processing jobs management page.""" + +import logging + +from fastapi import HTTPException, Request, status +from fastapi.responses import RedirectResponse + +from app.views.base import APIRouter, require_login, settings, templates + +logger = logging.getLogger(__name__) +router = APIRouter() + + +def _require_admin(request: Request): + """Return the session user if they are an admin, else None.""" + user = request.session.get("user") + if not user or not user.get("is_admin"): + logger.warning("Non-admin user attempted to access /admin/scheduled-jobs") + return None + return user + + +@router.get("/admin/scheduled-jobs") +@require_login +async def scheduled_jobs_page(request: Request): + """Admin scheduled jobs management page.""" + user = _require_admin(request) + if user is None: + return RedirectResponse(url="/", status_code=status.HTTP_302_FOUND) + + try: + return templates.TemplateResponse( + "admin_scheduled_jobs.html", + { + "request": request, + "app_version": settings.version, + }, + ) + except Exception as e: + logger.error(f"Error loading scheduled jobs page: {e}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to load scheduled jobs page", + ) diff --git a/frontend/templates/admin_scheduled_jobs.html b/frontend/templates/admin_scheduled_jobs.html new file mode 100644 index 00000000..cf4ed93d --- /dev/null +++ b/frontend/templates/admin_scheduled_jobs.html @@ -0,0 +1,558 @@ +{% extends "base.html" %} +{% block title %}Scheduled Jobs – Admin – DocuElevate{% endblock %} + +{% block content %} +
+ + +
+
+

+ + Scheduled Jobs + Admin Only +

+

+ Manage and trigger scheduled batch processing jobs. Schedule changes take effect after the worker restarts. +

+
+ +
+ + + + + +
+ +
+ How scheduling works: + These jobs are executed by the Celery Beat scheduler running inside the worker container. + Schedule changes (cron / interval) and enable / disable toggles are persisted immediately, + but the Celery Beat process reads the schedule only at startup — so changes take effect + after the worker is restarted. You can always trigger any job right now + using the Run Now button without restarting. +
+
+ + +
+
+ Loading… +
+
+ + +
+ + + +
+ + + + +
+{% endblock %} + +{% block scripts %} + +{% endblock %} diff --git a/migrations/versions/026_add_scheduled_jobs.py b/migrations/versions/026_add_scheduled_jobs.py new file mode 100644 index 00000000..a8e032b8 --- /dev/null +++ b/migrations/versions/026_add_scheduled_jobs.py @@ -0,0 +1,51 @@ +"""Add scheduled_jobs table for admin-managed batch processing schedules. + +Revision ID: 026_add_scheduled_jobs +Revises: 025_add_user_notifications +Create Date: 2026-03-09 +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "026_add_scheduled_jobs" +down_revision: Union[str, None] = "025_add_user_notifications" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create scheduled_jobs table.""" + op.create_table( + "scheduled_jobs", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(100), nullable=False), + sa.Column("display_name", sa.String(255), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("task_name", sa.String(255), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False, server_default="1"), + sa.Column("schedule_type", sa.String(20), nullable=False, server_default="cron"), + sa.Column("cron_minute", sa.String(50), nullable=False, server_default="0"), + sa.Column("cron_hour", sa.String(50), nullable=False, server_default="*"), + sa.Column("cron_day_of_week", sa.String(50), nullable=False, server_default="*"), + sa.Column("cron_day_of_month", sa.String(50), nullable=False, server_default="*"), + sa.Column("cron_month_of_year", sa.String(50), nullable=False, server_default="*"), + sa.Column("interval_seconds", sa.Integer(), nullable=True), + sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_run_status", sa.String(20), nullable=True), + sa.Column("last_run_detail", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("name", name="uq_scheduled_jobs_name"), + ) + op.create_index("ix_scheduled_jobs_id", "scheduled_jobs", ["id"]) + op.create_index("ix_scheduled_jobs_name", "scheduled_jobs", ["name"]) + + +def downgrade() -> None: + """Drop scheduled_jobs table.""" + op.drop_index("ix_scheduled_jobs_name", "scheduled_jobs") + op.drop_index("ix_scheduled_jobs_id", "scheduled_jobs") + op.drop_table("scheduled_jobs")