Merge pull request #576 from christianlouis/copilot/add-scheduled-batch-processing
feat: scheduled batch processing — 8 admin-managed jobs, DB-driven beat schedule, 100% coverage
This commit is contained in:
Christian Krakau-Louis
@@ -29,6 +29,7 @@ from app.api.plans import router as plans_router
|
||||
from app.api.process import router as process_router
|
||||
from app.api.queue import router as queue_router
|
||||
from app.api.saved_searches import router as saved_searches_router
|
||||
from app.api.scheduled_jobs import router as scheduled_jobs_router
|
||||
from app.api.search import router as search_router
|
||||
from app.api.settings import router as settings_router
|
||||
from app.api.shared_links import public_router as shared_links_public_router
|
||||
@@ -80,3 +81,4 @@ router.include_router(pipelines_router)
|
||||
router.include_router(imap_accounts_router)
|
||||
router.include_router(integrations_router)
|
||||
router.include_router(notifications_router)
|
||||
router.include_router(scheduled_jobs_router)
|
||||
|
||||
@@ -0,0 +1,350 @@
|
||||
"""
|
||||
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
|
||||
|
||||
model_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,
|
||||
},
|
||||
{
|
||||
"name": "expire-shared-links",
|
||||
"display_name": "Expire Stale Shared Links",
|
||||
"description": (
|
||||
"Marks shared document links as inactive when their expiry time has passed. "
|
||||
"Access is already blocked at request time, but this task keeps the "
|
||||
"management UI counts accurate. Runs daily at 01:00 UTC by default."
|
||||
),
|
||||
"task_name": "app.tasks.batch_tasks.expire_shared_links",
|
||||
"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": "prune-processing-logs",
|
||||
"display_name": "Prune Old Processing Logs",
|
||||
"description": (
|
||||
"Deletes processing log entries and settings audit log entries older than "
|
||||
"30 days to prevent unbounded database growth. "
|
||||
"Runs weekly on Sunday at 04:00 UTC by default."
|
||||
),
|
||||
"task_name": "app.tasks.batch_tasks.prune_processing_logs",
|
||||
"enabled": True,
|
||||
"schedule_type": "cron",
|
||||
"cron_minute": "0",
|
||||
"cron_hour": "4",
|
||||
"cron_day_of_week": "0",
|
||||
"cron_day_of_month": "*",
|
||||
"cron_month_of_year": "*",
|
||||
"interval_seconds": None,
|
||||
},
|
||||
{
|
||||
"name": "prune-old-notifications",
|
||||
"display_name": "Prune Old Notifications",
|
||||
"description": (
|
||||
"Deletes read in-app notifications older than 30 days. "
|
||||
"Unread notifications are never deleted. "
|
||||
"Runs weekly on Sunday at 04:30 UTC by default."
|
||||
),
|
||||
"task_name": "app.tasks.batch_tasks.prune_old_notifications",
|
||||
"enabled": True,
|
||||
"schedule_type": "cron",
|
||||
"cron_minute": "30",
|
||||
"cron_hour": "4",
|
||||
"cron_day_of_week": "0",
|
||||
"cron_day_of_month": "*",
|
||||
"cron_month_of_year": "*",
|
||||
"interval_seconds": None,
|
||||
},
|
||||
{
|
||||
"name": "backfill-missing-metadata",
|
||||
"display_name": "Backfill Missing AI Metadata",
|
||||
"description": (
|
||||
"Re-triggers AI metadata extraction for documents that have extracted "
|
||||
"text but no AI metadata yet (e.g., processed before an AI provider "
|
||||
"was configured). Processes up to 50 documents per run. "
|
||||
"Runs every 6 hours by default."
|
||||
),
|
||||
"task_name": "app.tasks.batch_tasks.backfill_missing_metadata",
|
||||
"enabled": True,
|
||||
"schedule_type": "cron",
|
||||
"cron_minute": "0",
|
||||
"cron_hour": "*/6",
|
||||
"cron_day_of_week": "*",
|
||||
"cron_day_of_month": "*",
|
||||
"cron_month_of_year": "*",
|
||||
"interval_seconds": None,
|
||||
},
|
||||
{
|
||||
"name": "sync-search-index",
|
||||
"display_name": "Sync Search Index",
|
||||
"description": (
|
||||
"Indexes documents that have OCR text or AI metadata but are missing "
|
||||
"from the Meilisearch search index. Useful after enabling search on "
|
||||
"an existing installation or after an index rebuild. "
|
||||
"Processes up to 100 documents per run. "
|
||||
"Runs hourly by default."
|
||||
),
|
||||
"task_name": "app.tasks.batch_tasks.sync_search_index",
|
||||
"enabled": True,
|
||||
"schedule_type": "cron",
|
||||
"cron_minute": "15",
|
||||
"cron_hour": "*/1",
|
||||
"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,
|
||||
}
|
||||
@@ -1,5 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import logging
|
||||
|
||||
from celery.schedules import crontab
|
||||
|
||||
# Ensure tasks are loaded
|
||||
@@ -9,6 +11,16 @@ 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
|
||||
backfill_missing_metadata,
|
||||
cleanup_temp_files,
|
||||
expire_shared_links,
|
||||
process_new_documents,
|
||||
prune_old_notifications,
|
||||
prune_processing_logs,
|
||||
reprocess_failed_documents,
|
||||
sync_search_index,
|
||||
)
|
||||
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
|
||||
@@ -48,6 +60,8 @@ from app.tasks.webhook_tasks import deliver_webhook_task # noqa: F401
|
||||
# Register the settings reload signal handler so workers pick up config changes
|
||||
from app.utils.settings_sync import register_settings_reload_signal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
register_settings_reload_signal()
|
||||
|
||||
celery.conf.task_routes = {
|
||||
@@ -159,3 +173,66 @@ celery.conf.beat_schedule = {
|
||||
|
||||
# Remove None entries from beat_schedule
|
||||
celery.conf.beat_schedule = {k: v for k, v in celery.conf.beat_schedule.items() if v is not None}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Load admin-managed scheduled jobs from the database
|
||||
# ---------------------------------------------------------------------------
|
||||
# These jobs are defined in the ``scheduled_jobs`` table (seeded by
|
||||
# ``app.api.scheduled_jobs.seed_default_scheduled_jobs``) and can be
|
||||
# enabled/disabled and rescheduled via the admin UI at /admin/scheduled-jobs.
|
||||
# The schedule is read once at worker startup; changes take effect after
|
||||
# the worker is restarted.
|
||||
|
||||
|
||||
def _load_db_scheduled_jobs() -> None:
|
||||
"""
|
||||
Extend ``celery.conf.beat_schedule`` with entries from the ``scheduled_jobs``
|
||||
database table.
|
||||
|
||||
Only rows with ``enabled=True`` are added. Rows whose ``name`` key
|
||||
already exists in the static schedule (defined above) are skipped so
|
||||
that hardcoded entries cannot be overridden accidentally.
|
||||
|
||||
Failures are logged as warnings and do not prevent the worker from
|
||||
starting.
|
||||
"""
|
||||
try:
|
||||
from app.database import SessionLocal
|
||||
from app.models import ScheduledJob
|
||||
|
||||
with SessionLocal() as db:
|
||||
jobs = db.query(ScheduledJob).filter(ScheduledJob.enabled.is_(True)).all()
|
||||
|
||||
added = 0
|
||||
for job in jobs:
|
||||
if job.name in celery.conf.beat_schedule:
|
||||
# Static entry takes precedence; skip silently.
|
||||
continue
|
||||
|
||||
if job.schedule_type == "interval" and job.interval_seconds:
|
||||
from celery.schedules import schedule as interval_schedule
|
||||
|
||||
sched = interval_schedule(run_every=job.interval_seconds)
|
||||
else:
|
||||
# Default to cron.
|
||||
sched = crontab(
|
||||
minute=job.cron_minute,
|
||||
hour=job.cron_hour,
|
||||
day_of_week=job.cron_day_of_week,
|
||||
day_of_month=job.cron_day_of_month,
|
||||
month_of_year=job.cron_month_of_year,
|
||||
)
|
||||
|
||||
celery.conf.beat_schedule[job.name] = {
|
||||
"task": job.task_name,
|
||||
"schedule": sched,
|
||||
"options": {"expires": 3600},
|
||||
}
|
||||
added += 1
|
||||
|
||||
logger.info("Loaded %d scheduled job(s) from database into Celery Beat.", added)
|
||||
except Exception as exc:
|
||||
logger.warning("Could not load scheduled jobs from database: %s", exc)
|
||||
|
||||
|
||||
_load_db_scheduled_jobs()
|
||||
|
||||
+14
@@ -132,6 +132,20 @@ async def lifespan(app: FastAPI):
|
||||
except Exception:
|
||||
logging.debug("Default pipeline seeding skipped — DB may not be ready yet") # noqa: S110
|
||||
|
||||
# Seed the default scheduled batch processing jobs so they appear in the
|
||||
# admin UI (/admin/scheduled-jobs) on first startup.
|
||||
try:
|
||||
from app.api.scheduled_jobs import seed_default_scheduled_jobs as _seed_jobs
|
||||
from app.database import SessionLocal as _SessionLocal # noqa: F811
|
||||
|
||||
_db_jobs = _SessionLocal()
|
||||
try:
|
||||
_seed_jobs(_db_jobs)
|
||||
finally:
|
||||
_db_jobs.close()
|
||||
except Exception:
|
||||
logging.debug("Scheduled jobs seeding skipped — DB may not be ready yet") # noqa: S110
|
||||
|
||||
# Application is now running
|
||||
yield
|
||||
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -0,0 +1,669 @@
|
||||
"""
|
||||
Scheduled batch processing tasks for DocuElevate.
|
||||
|
||||
This module provides Celery tasks that can be scheduled via Celery Beat
|
||||
and managed through the admin UI (``/admin/scheduled-jobs``):
|
||||
|
||||
Core batch jobs
|
||||
---------------
|
||||
- ``process_new_documents`` – Queue 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.
|
||||
|
||||
Maintenance / housekeeping jobs
|
||||
--------------------------------
|
||||
- ``expire_shared_links`` – Auto-revoke SharedLinks whose ``expires_at`` has passed.
|
||||
- ``prune_processing_logs`` – Delete old rows from ``processing_logs`` and
|
||||
``settings_audit_log`` to prevent unbounded table growth.
|
||||
- ``prune_old_notifications`` – Delete old read ``in_app_notifications`` rows.
|
||||
- ``backfill_missing_metadata`` – Re-trigger AI metadata extraction for completed files
|
||||
that have OCR text but no ``ai_metadata``.
|
||||
- ``sync_search_index`` – Index documents in Meilisearch that have OCR text /
|
||||
metadata but are not yet in the search index.
|
||||
|
||||
Each task records its execution result back to the ``ScheduledJob`` table so
|
||||
the admin UI can display last-run times and statuses.
|
||||
"""
|
||||
|
||||
import json
|
||||
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,
|
||||
InAppNotification,
|
||||
ProcessingLog,
|
||||
ScheduledJob,
|
||||
SettingsAuditLog,
|
||||
SharedLink,
|
||||
)
|
||||
|
||||
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: # pragma: no cover – only reachable if file vanishes between iterdir() and stat()
|
||||
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); skipped {skipped} (too new or protected); {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)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task: expire stale shared links
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@celery.task(name="app.tasks.batch_tasks.expire_shared_links")
|
||||
def expire_shared_links() -> dict:
|
||||
"""
|
||||
Auto-revoke SharedLinks whose ``expires_at`` timestamp has passed.
|
||||
|
||||
The ``_is_link_valid`` helper in the shared-links API already blocks
|
||||
access at request time, but the database rows remain flagged as
|
||||
``is_active=True``. This task sweeps those rows and sets
|
||||
``is_active=False`` + ``revoked_at`` so the management UI reflects
|
||||
the true state and counts are accurate.
|
||||
|
||||
Returns a summary dict with ``revoked`` count.
|
||||
"""
|
||||
job_name = "expire-shared-links"
|
||||
logger.info("[batch] Starting expire_shared_links task")
|
||||
|
||||
try:
|
||||
now = datetime.now(timezone.utc)
|
||||
with SessionLocal() as db:
|
||||
stale = (
|
||||
db.query(SharedLink)
|
||||
.filter(
|
||||
SharedLink.is_active.is_(True),
|
||||
SharedLink.expires_at.isnot(None),
|
||||
SharedLink.expires_at < now,
|
||||
)
|
||||
.all()
|
||||
)
|
||||
for link in stale:
|
||||
link.is_active = False
|
||||
link.revoked_at = now
|
||||
db.commit()
|
||||
revoked = len(stale)
|
||||
|
||||
detail = f"Revoked {revoked} expired shared link(s)."
|
||||
logger.info("[batch] expire_shared_links: %s", detail)
|
||||
_update_job_status(job_name, "success", detail)
|
||||
return {"revoked": revoked}
|
||||
|
||||
except Exception as exc:
|
||||
detail = f"Error: {exc}"
|
||||
logger.error("[batch] expire_shared_links failed: %s", exc, exc_info=True)
|
||||
_update_job_status(job_name, "failed", detail)
|
||||
return {"revoked": 0, "error": str(exc)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task: prune old processing logs
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
#: Default retention period for processing logs and audit log rows.
|
||||
_LOG_RETENTION_DAYS: int = 30
|
||||
|
||||
|
||||
@celery.task(name="app.tasks.batch_tasks.prune_processing_logs")
|
||||
def prune_processing_logs(retention_days: int = _LOG_RETENTION_DAYS) -> dict:
|
||||
"""
|
||||
Delete ``processing_logs`` and ``settings_audit_log`` rows older than
|
||||
*retention_days* (default 30) to prevent unbounded table growth.
|
||||
|
||||
Rows for the most recent *retention_days* days are kept so that recent
|
||||
activity is still visible in the logs/audit UI.
|
||||
|
||||
Args:
|
||||
retention_days: Number of days of history to keep (default 30).
|
||||
|
||||
Returns:
|
||||
A summary dict with ``processing_logs_deleted`` and
|
||||
``audit_log_deleted`` counts.
|
||||
"""
|
||||
job_name = "prune-processing-logs"
|
||||
logger.info("[batch] Starting prune_processing_logs (retention_days=%s)", retention_days)
|
||||
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
|
||||
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
pl_deleted = db.query(ProcessingLog).filter(ProcessingLog.timestamp < cutoff).delete()
|
||||
al_deleted = db.query(SettingsAuditLog).filter(SettingsAuditLog.changed_at < cutoff).delete()
|
||||
db.commit()
|
||||
|
||||
detail = (
|
||||
f"Deleted {pl_deleted} processing log row(s) and "
|
||||
f"{al_deleted} settings audit log row(s) older than {retention_days} days."
|
||||
)
|
||||
logger.info("[batch] prune_processing_logs: %s", detail)
|
||||
_update_job_status(job_name, "success", detail)
|
||||
return {"processing_logs_deleted": pl_deleted, "audit_log_deleted": al_deleted}
|
||||
|
||||
except Exception as exc:
|
||||
detail = f"Error: {exc}"
|
||||
logger.error("[batch] prune_processing_logs failed: %s", exc, exc_info=True)
|
||||
_update_job_status(job_name, "failed", detail)
|
||||
return {"processing_logs_deleted": 0, "audit_log_deleted": 0, "error": str(exc)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task: prune old in-app notifications
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
#: Default retention period for read notifications.
|
||||
_NOTIFICATION_RETENTION_DAYS: int = 30
|
||||
|
||||
|
||||
@celery.task(name="app.tasks.batch_tasks.prune_old_notifications")
|
||||
def prune_old_notifications(retention_days: int = _NOTIFICATION_RETENTION_DAYS) -> dict:
|
||||
"""
|
||||
Delete ``in_app_notifications`` rows that are already read and older than
|
||||
*retention_days* days (default 30) to prevent unbounded table growth.
|
||||
|
||||
Unread notifications are always kept regardless of age so users do not
|
||||
miss important alerts.
|
||||
|
||||
Args:
|
||||
retention_days: Number of days of read-notification history to keep
|
||||
(default 30).
|
||||
|
||||
Returns:
|
||||
A summary dict with ``deleted`` count.
|
||||
"""
|
||||
job_name = "prune-old-notifications"
|
||||
logger.info("[batch] Starting prune_old_notifications (retention_days=%s)", retention_days)
|
||||
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
|
||||
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
deleted = (
|
||||
db.query(InAppNotification)
|
||||
.filter(
|
||||
InAppNotification.is_read.is_(True),
|
||||
InAppNotification.created_at < cutoff,
|
||||
)
|
||||
.delete()
|
||||
)
|
||||
db.commit()
|
||||
|
||||
detail = f"Deleted {deleted} old read notification(s) older than {retention_days} days."
|
||||
logger.info("[batch] prune_old_notifications: %s", detail)
|
||||
_update_job_status(job_name, "success", detail)
|
||||
return {"deleted": deleted}
|
||||
|
||||
except Exception as exc:
|
||||
detail = f"Error: {exc}"
|
||||
logger.error("[batch] prune_old_notifications failed: %s", exc, exc_info=True)
|
||||
_update_job_status(job_name, "failed", detail)
|
||||
return {"deleted": 0, "error": str(exc)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task: backfill missing AI metadata
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
#: Maximum number of files to process per backfill run.
|
||||
_METADATA_BACKFILL_BATCH_SIZE: int = 50
|
||||
|
||||
|
||||
@celery.task(name="app.tasks.batch_tasks.backfill_missing_metadata")
|
||||
def backfill_missing_metadata(batch_size: int = _METADATA_BACKFILL_BATCH_SIZE) -> dict:
|
||||
"""
|
||||
Re-trigger AI metadata extraction for documents that have OCR text but
|
||||
no ``ai_metadata``.
|
||||
|
||||
This handles the common case where a document was processed before the AI
|
||||
metadata extraction step was configured (e.g., before an OpenAI API key
|
||||
was added), or where the extraction previously failed.
|
||||
|
||||
Only files that are **not** currently in-progress and have non-empty
|
||||
``ocr_text`` are selected. A configurable *batch_size* caps the number
|
||||
of tasks queued per run to avoid overwhelming the AI provider.
|
||||
|
||||
Args:
|
||||
batch_size: Maximum number of files to queue per run (default 50).
|
||||
|
||||
Returns:
|
||||
A summary dict with ``queued`` count.
|
||||
"""
|
||||
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt # avoid circular import
|
||||
|
||||
job_name = "backfill-missing-metadata"
|
||||
logger.info("[batch] Starting backfill_missing_metadata (batch_size=%s)", batch_size)
|
||||
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
# Files currently being processed — skip them.
|
||||
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.ocr_text.isnot(None))
|
||||
.filter(FileRecord.ocr_text != "")
|
||||
.filter((FileRecord.ai_metadata.is_(None)) | (FileRecord.ai_metadata == ""))
|
||||
.filter(~FileRecord.id.in_(db.query(in_progress_file_ids.c.file_id)))
|
||||
.limit(batch_size)
|
||||
.all()
|
||||
)
|
||||
|
||||
queued = 0
|
||||
for record in candidates:
|
||||
filename = record.local_filename or record.original_filename or f"file_{record.id}"
|
||||
extract_metadata_with_gpt.delay(
|
||||
filename,
|
||||
record.ocr_text,
|
||||
file_id=record.id,
|
||||
)
|
||||
queued += 1
|
||||
|
||||
detail = f"Queued {queued} document(s) for AI metadata backfill."
|
||||
logger.info("[batch] backfill_missing_metadata: %s", detail)
|
||||
_update_job_status(job_name, "success", detail)
|
||||
return {"queued": queued}
|
||||
|
||||
except Exception as exc:
|
||||
detail = f"Error: {exc}"
|
||||
logger.error("[batch] backfill_missing_metadata failed: %s", exc, exc_info=True)
|
||||
_update_job_status(job_name, "failed", detail)
|
||||
return {"queued": 0, "error": str(exc)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task: sync Meilisearch search index
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
#: Maximum documents to index per sync run.
|
||||
_SEARCH_SYNC_BATCH_SIZE: int = 100
|
||||
|
||||
|
||||
@celery.task(name="app.tasks.batch_tasks.sync_search_index")
|
||||
def sync_search_index(batch_size: int = _SEARCH_SYNC_BATCH_SIZE) -> dict:
|
||||
"""
|
||||
Index documents in Meilisearch that have OCR text or AI metadata but are
|
||||
not yet present in the search index.
|
||||
|
||||
This is useful after:
|
||||
- Enabling Meilisearch for the first time on an existing installation.
|
||||
- Recovering from a Meilisearch index wipe or migration.
|
||||
- Documents processed before search indexing was added to the pipeline.
|
||||
|
||||
The task queries the Meilisearch index for existing document IDs, then
|
||||
finds ``FileRecord`` rows that have processable content (``ocr_text`` or
|
||||
``ai_metadata``) but are absent from the index, and re-indexes them.
|
||||
|
||||
A configurable *batch_size* caps the number of documents indexed per run.
|
||||
|
||||
Args:
|
||||
batch_size: Maximum number of documents to index per run (default 100).
|
||||
|
||||
Returns:
|
||||
A summary dict with ``indexed`` and ``skipped`` counts.
|
||||
"""
|
||||
from app.utils.meilisearch_client import get_meilisearch_client, index_document
|
||||
|
||||
job_name = "sync-search-index"
|
||||
logger.info("[batch] Starting sync_search_index (batch_size=%s)", batch_size)
|
||||
|
||||
client = get_meilisearch_client()
|
||||
if client is None:
|
||||
detail = "Meilisearch is not configured; skipping search index sync."
|
||||
logger.info("[batch] sync_search_index: %s", detail)
|
||||
_update_job_status(job_name, "success", detail)
|
||||
return {"indexed": 0, "skipped": 0, "reason": "meilisearch_not_configured"}
|
||||
|
||||
try:
|
||||
# Fetch the set of file_ids already in the Meilisearch index.
|
||||
index = client.get_index(settings.meilisearch_index_name)
|
||||
# Fetch up to 10 000 IDs — sufficient to determine gaps for most installs.
|
||||
existing_result = index.get_documents({"fields": ["file_id"], "limit": 10000})
|
||||
existing_ids: set[int] = {doc["file_id"] for doc in existing_result.results if "file_id" in doc}
|
||||
except Exception as exc:
|
||||
detail = f"Error fetching existing Meilisearch IDs: {exc}"
|
||||
logger.error("[batch] sync_search_index: %s", detail)
|
||||
_update_job_status(job_name, "failed", detail)
|
||||
return {"indexed": 0, "skipped": 0, "error": str(exc)}
|
||||
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
# Files with indexable content that are not already in the index.
|
||||
candidates = (
|
||||
db.query(FileRecord)
|
||||
.filter(FileRecord.is_duplicate.is_(False))
|
||||
.filter(
|
||||
(FileRecord.ocr_text.isnot(None) & (FileRecord.ocr_text != ""))
|
||||
| (FileRecord.ai_metadata.isnot(None) & (FileRecord.ai_metadata != ""))
|
||||
)
|
||||
.filter(~FileRecord.id.in_(existing_ids) if existing_ids else True) # type: ignore[arg-type]
|
||||
.limit(batch_size)
|
||||
.all()
|
||||
)
|
||||
|
||||
indexed = 0
|
||||
skipped = 0
|
||||
for record in candidates:
|
||||
metadata: dict = {}
|
||||
if record.ai_metadata:
|
||||
try:
|
||||
metadata = json.loads(record.ai_metadata)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
pass
|
||||
|
||||
success = index_document(record, record.ocr_text or "", metadata)
|
||||
if success:
|
||||
indexed += 1
|
||||
else:
|
||||
skipped += 1
|
||||
|
||||
detail = f"Indexed {indexed} document(s) into Meilisearch; {skipped} skipped (indexing error)."
|
||||
logger.info("[batch] sync_search_index: %s", detail)
|
||||
_update_job_status(job_name, "success", detail)
|
||||
return {"indexed": indexed, "skipped": skipped}
|
||||
|
||||
except Exception as exc:
|
||||
detail = f"Error: {exc}"
|
||||
logger.error("[batch] sync_search_index failed: %s", exc, exc_info=True)
|
||||
_update_job_status(job_name, "failed", detail)
|
||||
return {"indexed": 0, "skipped": 0, "error": str(exc)}
|
||||
@@ -24,6 +24,7 @@ from app.views.onedrive import router as onedrive_router
|
||||
from app.views.pipelines import router as pipelines_router # Processing pipelines
|
||||
from app.views.plans import router as plans_router # Admin Plan Designer
|
||||
from app.views.queue import router as queue_router
|
||||
from app.views.scheduled_jobs import router as scheduled_jobs_router # Scheduled batch jobs
|
||||
from app.views.search import router as search_router
|
||||
from app.views.settings import router as settings_router
|
||||
from app.views.share import router as share_router
|
||||
@@ -58,4 +59,5 @@ router.include_router(pipelines_router) # Processing pipelines
|
||||
router.include_router(imap_accounts_router) # Per-user IMAP ingestion accounts
|
||||
router.include_router(integrations_router) # Unified integrations dashboard
|
||||
router.include_router(notifications_router) # User notification dashboard
|
||||
router.include_router(scheduled_jobs_router) # Admin scheduled batch jobs
|
||||
router.include_router(help_router) # Built-in help / How-To docs
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
@@ -0,0 +1,226 @@
|
||||
# Scheduled Jobs
|
||||
|
||||
DocuElevate includes an admin-managed **Scheduled Jobs** system that runs
|
||||
recurring maintenance and processing tasks automatically via Celery Beat.
|
||||
|
||||
All jobs are visible and configurable through the admin UI at
|
||||
**Admin → Scheduled Jobs** (`/admin/scheduled-jobs`).
|
||||
|
||||
---
|
||||
|
||||
## Overview
|
||||
|
||||
Scheduled jobs replace the need to run manual batch operations. Each job:
|
||||
|
||||
- Runs automatically on its configured cron or interval schedule.
|
||||
- Can be **enabled** or **disabled** without restarting the worker.
|
||||
- Can be triggered **immediately** with the *Run Now* button.
|
||||
- Reports its last-run time, status (`success`, `failed`, `running`), and
|
||||
a short detail message back to the UI.
|
||||
|
||||
> **Note:** Schedule changes (cron expressions, interval values, enable/disable
|
||||
> toggles) are persisted to the database immediately. However, Celery Beat
|
||||
> reads the schedule only at worker startup — so changes take effect after
|
||||
> **restarting the Celery worker**.
|
||||
> The *Run Now* button dispatches a job immediately and does **not** require
|
||||
> a restart.
|
||||
|
||||
---
|
||||
|
||||
## Built-in Jobs
|
||||
|
||||
### 1. Process New Documents
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.process_new_documents` |
|
||||
| **Default schedule** | Every hour (cron `0 */1 * * *`) |
|
||||
| **Purpose** | Scans for documents that have been uploaded but never processed, then queues them through the full pipeline. |
|
||||
|
||||
A document is considered *new* when it has no `FileProcessingStep` rows
|
||||
matching the core pipeline steps. Only files whose `local_filename` still
|
||||
exists on disk are queued; others are skipped and counted.
|
||||
|
||||
---
|
||||
|
||||
### 2. Reprocess Failed Documents
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.reprocess_failed_documents` |
|
||||
| **Default schedule** | Every 6 hours (cron `30 */6 * * *`) |
|
||||
| **Purpose** | Finds documents whose last processing step has status `failure` and re-queues them. Skips files that are currently in-progress. |
|
||||
|
||||
---
|
||||
|
||||
### 3. Clean Up Temporary Files
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.cleanup_temp_files` |
|
||||
| **Default schedule** | Daily at 03:30 UTC (cron `30 3 * * *`) |
|
||||
| **Purpose** | Removes stale files from `workdir/tmp`. Only files older than 24 hours that are not referenced by any active processing job are deleted. |
|
||||
|
||||
---
|
||||
|
||||
### 4. Expire Stale Shared Links
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.expire_shared_links` |
|
||||
| **Default schedule** | Daily at 01:00 UTC (cron `0 1 * * *`) |
|
||||
| **Purpose** | Marks shared document links as inactive when their `expires_at` timestamp has passed. |
|
||||
|
||||
Access is already blocked at request time, but this task keeps the management
|
||||
UI counts and statuses accurate.
|
||||
|
||||
---
|
||||
|
||||
### 5. Prune Old Processing Logs
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.prune_processing_logs` |
|
||||
| **Default schedule** | Weekly on Sunday at 04:00 UTC (cron `0 4 * * 0`) |
|
||||
| **Purpose** | Deletes `processing_logs` and `settings_audit_log` rows older than 30 days to prevent unbounded table growth. |
|
||||
|
||||
---
|
||||
|
||||
### 6. Prune Old Notifications
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.prune_old_notifications` |
|
||||
| **Default schedule** | Weekly on Sunday at 04:30 UTC (cron `30 4 * * 0`) |
|
||||
| **Purpose** | Deletes *read* `in_app_notifications` rows older than 30 days. Unread notifications are never deleted. |
|
||||
|
||||
---
|
||||
|
||||
### 7. Backfill Missing AI Metadata
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.backfill_missing_metadata` |
|
||||
| **Default schedule** | Every 6 hours (cron `0 */6 * * *`) |
|
||||
| **Purpose** | Re-triggers AI metadata extraction for documents that have OCR text but no `ai_metadata`. Processes up to 50 documents per run. |
|
||||
|
||||
This is particularly useful when:
|
||||
- An AI provider (OpenAI / Azure) was configured after documents were already
|
||||
ingested.
|
||||
- A previous metadata extraction attempt failed.
|
||||
|
||||
---
|
||||
|
||||
### 8. Sync Search Index
|
||||
|
||||
| Field | Value |
|
||||
|---|---|
|
||||
| **Task** | `app.tasks.batch_tasks.sync_search_index` |
|
||||
| **Default schedule** | Hourly (cron `15 */1 * * *`) |
|
||||
| **Purpose** | Indexes documents in Meilisearch that have OCR text or AI metadata but are not yet present in the search index. Processes up to 100 documents per run. |
|
||||
|
||||
This is useful after:
|
||||
- Enabling Meilisearch for the first time on an existing installation.
|
||||
- Recovering from a Meilisearch index wipe or migration.
|
||||
|
||||
---
|
||||
|
||||
## Managing Schedules
|
||||
|
||||
### Editing a schedule
|
||||
|
||||
1. Navigate to **Admin → Scheduled Jobs**.
|
||||
2. Click **Edit Schedule** on the job card.
|
||||
3. Choose **Cron** or **Interval** and fill in the fields.
|
||||
4. Click **Save Changes**.
|
||||
|
||||
**Restart the worker** for the new schedule to take effect:
|
||||
|
||||
```bash
|
||||
# Docker Compose
|
||||
docker compose restart worker
|
||||
|
||||
# Kubernetes
|
||||
kubectl rollout restart deployment/docuelevate-worker
|
||||
```
|
||||
|
||||
### Cron format
|
||||
|
||||
Cron expressions follow the standard 5-field format:
|
||||
|
||||
```
|
||||
minute hour day-of-month month day-of-week
|
||||
```
|
||||
|
||||
Examples:
|
||||
|
||||
| Expression | Meaning |
|
||||
|---|---|
|
||||
| `0 * * * *` | Every hour on the hour |
|
||||
| `0 2 * * *` | Daily at 02:00 UTC |
|
||||
| `30 4 * * 0` | Every Sunday at 04:30 UTC |
|
||||
| `*/15 * * * *` | Every 15 minutes |
|
||||
|
||||
### Interval format
|
||||
|
||||
For interval schedules, provide the number of **seconds**:
|
||||
|
||||
| Seconds | Meaning |
|
||||
|---|---|
|
||||
| `60` | Every minute |
|
||||
| `3600` | Every hour |
|
||||
| `86400` | Every day |
|
||||
|
||||
The minimum interval is **60 seconds**.
|
||||
|
||||
---
|
||||
|
||||
## Enabling / Disabling a job
|
||||
|
||||
Click the toggle on the job card to enable or disable the job. The change
|
||||
is saved immediately, but the updated schedule only takes effect after the
|
||||
worker restarts.
|
||||
|
||||
Disabling a job does **not** cancel an in-flight execution — it only prevents
|
||||
Celery Beat from scheduling new executions.
|
||||
|
||||
---
|
||||
|
||||
## Run Now
|
||||
|
||||
Click **Run Now** on any job card to dispatch it immediately, regardless of
|
||||
its schedule or enabled status. The task is sent to the `default` Celery
|
||||
queue and will be picked up by the next available worker.
|
||||
|
||||
The UI shows a spinner while the dispatch request is in-flight and displays
|
||||
the Celery task ID on success. Refresh the page after a few seconds to see
|
||||
the updated last-run status.
|
||||
|
||||
---
|
||||
|
||||
## Adding custom jobs
|
||||
|
||||
Custom batch jobs can be added by:
|
||||
|
||||
1. Creating a new Celery task in `app/tasks/` (or any existing tasks module).
|
||||
2. Adding a row to `DEFAULT_JOBS` in `app/api/scheduled_jobs.py`.
|
||||
3. Running the app — the new job will be seeded automatically on startup.
|
||||
|
||||
The task function should call `_update_job_status(job_name, status, detail)`
|
||||
at the end of its execution so the admin UI reflects the outcome.
|
||||
|
||||
---
|
||||
|
||||
## Architecture notes
|
||||
|
||||
- **Seeding**: Default jobs are seeded into the `scheduled_jobs` database
|
||||
table when the FastAPI application starts up (via the lifespan handler in
|
||||
`app/main.py`). Seeding is idempotent — re-running does not create
|
||||
duplicate rows.
|
||||
- **Beat integration**: `app/celery_worker.py` calls `_load_db_scheduled_jobs()`
|
||||
at import time to extend `celery.conf.beat_schedule` with enabled DB jobs.
|
||||
Static hardcoded entries always take precedence over DB entries with the
|
||||
same key.
|
||||
- **Status tracking**: Each task calls `_update_job_status()` in
|
||||
`app/tasks/batch_tasks.py` to persist `last_run_at`, `last_run_status`, and
|
||||
`last_run_detail` back to the `scheduled_jobs` table.
|
||||
@@ -0,0 +1,558 @@
|
||||
{% extends "base.html" %}
|
||||
{% block title %}Scheduled Jobs – Admin – DocuElevate{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<div class="container mx-auto px-4 py-8" x-data="scheduledJobsApp()">
|
||||
|
||||
<!-- ── Header ─────────────────────────────────────────────────────────────── -->
|
||||
<div class="mb-6 flex flex-col sm:flex-row sm:items-center sm:justify-between gap-3">
|
||||
<div>
|
||||
<h1 class="text-3xl font-bold text-gray-900 flex items-center gap-2">
|
||||
<i class="fas fa-clock text-indigo-500" aria-hidden="true"></i>
|
||||
Scheduled Jobs
|
||||
<span class="text-xs font-semibold text-red-600 bg-red-50 border border-red-200 rounded px-2 py-0.5 ml-1">Admin Only</span>
|
||||
</h1>
|
||||
<p class="text-gray-500 text-sm mt-1">
|
||||
Manage and trigger scheduled batch processing jobs. Schedule changes take effect after the worker restarts.
|
||||
</p>
|
||||
</div>
|
||||
<button
|
||||
type="button"
|
||||
@click="fetchJobs()"
|
||||
class="inline-flex items-center px-4 py-2 bg-gray-100 hover:bg-gray-200 text-gray-700 text-sm font-medium rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-gray-400"
|
||||
aria-label="Refresh job list"
|
||||
>
|
||||
<i class="fas fa-sync-alt mr-2" aria-hidden="true"></i> Refresh
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<!-- ── Alert banner ───────────────────────────────────────────────────────── -->
|
||||
<div x-show="alert.show" x-transition class="mb-4" role="alert" :aria-live="alert.type === 'error' ? 'assertive' : 'polite'">
|
||||
<div
|
||||
:class="alert.type === 'success'
|
||||
? 'bg-green-50 border-green-400 text-green-800'
|
||||
: 'bg-red-50 border-red-400 text-red-800'"
|
||||
class="border-l-4 p-4 rounded"
|
||||
>
|
||||
<p class="font-semibold" x-text="alert.title"></p>
|
||||
<p class="text-sm mt-1" x-text="alert.message"></p>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- ── Info box ───────────────────────────────────────────────────────────── -->
|
||||
<div class="mb-6 bg-blue-50 border border-blue-200 rounded-lg p-4 text-sm text-blue-800 flex items-start gap-3">
|
||||
<i class="fas fa-info-circle mt-0.5 flex-shrink-0" aria-hidden="true"></i>
|
||||
<div>
|
||||
<strong>How scheduling works:</strong>
|
||||
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 <strong>right now</strong>
|
||||
using the <em>Run Now</em> button without restarting.
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- ── Loading spinner ────────────────────────────────────────────────────── -->
|
||||
<div x-show="loading" class="flex justify-center py-12" aria-live="polite" aria-label="Loading scheduled jobs">
|
||||
<div class="animate-spin rounded-full h-8 w-8 border-b-2 border-indigo-600" role="status">
|
||||
<span class="sr-only">Loading…</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- ── Jobs list ──────────────────────────────────────────────────────────── -->
|
||||
<div x-show="!loading" class="space-y-4">
|
||||
<template x-if="jobs.length === 0">
|
||||
<div class="text-center py-12 text-gray-500">
|
||||
<i class="fas fa-clock text-4xl mb-3 opacity-30" aria-hidden="true"></i>
|
||||
<p>No scheduled jobs found.</p>
|
||||
</div>
|
||||
</template>
|
||||
|
||||
<template x-for="job in jobs" :key="job.id">
|
||||
<div class="bg-white shadow rounded-lg overflow-hidden">
|
||||
<!-- Card header -->
|
||||
<div class="px-6 py-4 flex flex-col sm:flex-row sm:items-center sm:justify-between gap-3 border-b border-gray-100">
|
||||
<div class="flex items-center gap-3">
|
||||
<!-- Enable / disable toggle -->
|
||||
<button
|
||||
type="button"
|
||||
@click="toggleEnabled(job)"
|
||||
:aria-label="job.enabled ? 'Disable job ' + job.display_name : 'Enable job ' + job.display_name"
|
||||
:aria-pressed="job.enabled"
|
||||
class="relative inline-flex h-6 w-11 flex-shrink-0 cursor-pointer rounded-full border-2 border-transparent transition-colors duration-200 ease-in-out focus:outline-none focus:ring-2 focus:ring-indigo-500 focus:ring-offset-2"
|
||||
:class="job.enabled ? 'bg-indigo-600' : 'bg-gray-200'"
|
||||
>
|
||||
<span
|
||||
class="pointer-events-none inline-block h-5 w-5 transform rounded-full bg-white shadow ring-0 transition duration-200 ease-in-out"
|
||||
:class="job.enabled ? 'translate-x-5' : 'translate-x-0'"
|
||||
></span>
|
||||
</button>
|
||||
<div>
|
||||
<h2 class="text-base font-semibold text-gray-900" x-text="job.display_name"></h2>
|
||||
<p class="text-xs text-gray-500 mt-0.5" x-text="job.description"></p>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="flex items-center gap-2 flex-wrap">
|
||||
<!-- Status badge -->
|
||||
<span
|
||||
class="inline-flex items-center px-2.5 py-0.5 rounded-full text-xs font-medium"
|
||||
:class="{
|
||||
'bg-green-100 text-green-800': job.enabled,
|
||||
'bg-gray-100 text-gray-600': !job.enabled
|
||||
}"
|
||||
>
|
||||
<span
|
||||
class="w-1.5 h-1.5 rounded-full mr-1.5"
|
||||
:class="job.enabled ? 'bg-green-500' : 'bg-gray-400'"
|
||||
></span>
|
||||
<span x-text="job.enabled ? 'Active' : 'Disabled'"></span>
|
||||
</span>
|
||||
|
||||
<!-- Run Now button -->
|
||||
<button
|
||||
type="button"
|
||||
@click="runNow(job)"
|
||||
:disabled="runningJobIds.includes(job.id)"
|
||||
class="inline-flex items-center px-3 py-1.5 bg-indigo-600 hover:bg-indigo-700 disabled:opacity-50 disabled:cursor-not-allowed text-white text-xs font-medium rounded-md focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
|
||||
:aria-label="'Run job ' + job.display_name + ' now'"
|
||||
>
|
||||
<template x-if="runningJobIds.includes(job.id)">
|
||||
<i class="fas fa-spinner fa-spin mr-1.5" aria-hidden="true"></i>
|
||||
</template>
|
||||
<template x-if="!runningJobIds.includes(job.id)">
|
||||
<i class="fas fa-play mr-1.5" aria-hidden="true"></i>
|
||||
</template>
|
||||
Run Now
|
||||
</button>
|
||||
|
||||
<!-- Edit button -->
|
||||
<button
|
||||
type="button"
|
||||
@click="openEditModal(job)"
|
||||
class="inline-flex items-center px-3 py-1.5 bg-white border border-gray-300 hover:bg-gray-50 text-gray-700 text-xs font-medium rounded-md focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
|
||||
:aria-label="'Edit schedule for ' + job.display_name"
|
||||
>
|
||||
<i class="fas fa-edit mr-1.5" aria-hidden="true"></i> Edit Schedule
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Card body: schedule info + last run -->
|
||||
<div class="px-6 py-4 grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4 text-sm">
|
||||
|
||||
<!-- Schedule type -->
|
||||
<div>
|
||||
<span class="block text-xs font-medium text-gray-400 uppercase tracking-wider mb-1">Schedule Type</span>
|
||||
<span
|
||||
class="inline-flex items-center gap-1 font-medium text-gray-700"
|
||||
x-text="job.schedule_type === 'cron' ? 'Cron' : 'Interval'"
|
||||
></span>
|
||||
</div>
|
||||
|
||||
<!-- Schedule expression -->
|
||||
<div>
|
||||
<span class="block text-xs font-medium text-gray-400 uppercase tracking-wider mb-1">Schedule</span>
|
||||
<code
|
||||
class="text-xs bg-gray-100 rounded px-1.5 py-0.5 font-mono text-gray-800"
|
||||
x-text="formatSchedule(job)"
|
||||
></code>
|
||||
</div>
|
||||
|
||||
<!-- Last run -->
|
||||
<div>
|
||||
<span class="block text-xs font-medium text-gray-400 uppercase tracking-wider mb-1">Last Run</span>
|
||||
<span
|
||||
class="text-gray-700"
|
||||
x-text="job.last_run_at ? formatDate(job.last_run_at) : 'Never'"
|
||||
></span>
|
||||
</div>
|
||||
|
||||
<!-- Last result -->
|
||||
<div>
|
||||
<span class="block text-xs font-medium text-gray-400 uppercase tracking-wider mb-1">Last Result</span>
|
||||
<template x-if="job.last_run_status">
|
||||
<span
|
||||
class="inline-flex items-center gap-1 px-2 py-0.5 rounded-full text-xs font-medium"
|
||||
:class="{
|
||||
'bg-green-100 text-green-800': job.last_run_status === 'success',
|
||||
'bg-red-100 text-red-800': job.last_run_status === 'failed',
|
||||
'bg-yellow-100 text-yellow-800': job.last_run_status === 'running'
|
||||
}"
|
||||
>
|
||||
<i
|
||||
:class="{
|
||||
'fas fa-check-circle': job.last_run_status === 'success',
|
||||
'fas fa-exclamation-circle': job.last_run_status === 'failed',
|
||||
'fas fa-spinner fa-spin': job.last_run_status === 'running'
|
||||
}"
|
||||
aria-hidden="true"
|
||||
></i>
|
||||
<span x-text="job.last_run_status.charAt(0).toUpperCase() + job.last_run_status.slice(1)"></span>
|
||||
</span>
|
||||
</template>
|
||||
<template x-if="!job.last_run_status">
|
||||
<span class="text-gray-400 text-xs">—</span>
|
||||
</template>
|
||||
<template x-if="job.last_run_detail">
|
||||
<p class="text-xs text-gray-500 mt-1" x-text="job.last_run_detail"></p>
|
||||
</template>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
|
||||
<!-- Task name (collapsed) -->
|
||||
<div class="px-6 pb-4">
|
||||
<details class="text-xs text-gray-400">
|
||||
<summary class="cursor-pointer hover:text-gray-600 select-none">Technical details</summary>
|
||||
<div class="mt-2 space-y-1">
|
||||
<p><span class="font-medium">Task:</span> <code class="bg-gray-100 rounded px-1" x-text="job.task_name"></code></p>
|
||||
<p><span class="font-medium">Job key:</span> <code class="bg-gray-100 rounded px-1" x-text="job.name"></code></p>
|
||||
</div>
|
||||
</details>
|
||||
</div>
|
||||
</div>
|
||||
</template>
|
||||
</div>
|
||||
|
||||
<!-- ── Edit schedule modal ────────────────────────────────────────────────── -->
|
||||
<div
|
||||
x-show="editModal.open"
|
||||
x-transition:enter="ease-out duration-200"
|
||||
x-transition:enter-start="opacity-0"
|
||||
x-transition:enter-end="opacity-100"
|
||||
x-transition:leave="ease-in duration-150"
|
||||
x-transition:leave-start="opacity-100"
|
||||
x-transition:leave-end="opacity-0"
|
||||
class="fixed inset-0 z-50 overflow-y-auto"
|
||||
role="dialog"
|
||||
aria-modal="true"
|
||||
:aria-labelledby="'edit-modal-title-' + (editModal.job ? editModal.job.id : '')"
|
||||
x-cloak
|
||||
>
|
||||
<div class="flex items-center justify-center min-h-screen px-4 py-8">
|
||||
<!-- Backdrop -->
|
||||
<div class="fixed inset-0 bg-gray-500 bg-opacity-75" @click="closeEditModal()" aria-hidden="true"></div>
|
||||
|
||||
<!-- Panel -->
|
||||
<div
|
||||
class="relative bg-white rounded-lg shadow-xl w-full max-w-lg p-6 z-10"
|
||||
@click.stop
|
||||
>
|
||||
<div class="mb-4 flex items-center justify-between">
|
||||
<h2
|
||||
:id="'edit-modal-title-' + (editModal.job ? editModal.job.id : '')"
|
||||
class="text-lg font-semibold text-gray-900"
|
||||
x-text="'Edit Schedule: ' + (editModal.job ? editModal.job.display_name : '')"
|
||||
></h2>
|
||||
<button
|
||||
type="button"
|
||||
@click="closeEditModal()"
|
||||
class="text-gray-400 hover:text-gray-600 focus:outline-none focus:ring-2 focus:ring-indigo-500 rounded"
|
||||
aria-label="Close edit modal"
|
||||
>
|
||||
<i class="fas fa-times" aria-hidden="true"></i>
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<form @submit.prevent="saveSchedule()" novalidate>
|
||||
|
||||
<!-- Schedule type -->
|
||||
<div class="mb-4">
|
||||
<label class="block text-sm font-medium text-gray-700 mb-1" for="scheduleType">Schedule Type</label>
|
||||
<select
|
||||
id="scheduleType"
|
||||
x-model="editModal.form.schedule_type"
|
||||
class="w-full border border-gray-300 rounded-md px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-indigo-400"
|
||||
>
|
||||
<option value="cron">Cron</option>
|
||||
<option value="interval">Interval</option>
|
||||
</select>
|
||||
</div>
|
||||
|
||||
<!-- Cron fields -->
|
||||
<div x-show="editModal.form.schedule_type === 'cron'" class="space-y-3 mb-4">
|
||||
<p class="text-xs text-gray-500">
|
||||
Standard cron expressions. Use <code class="bg-gray-100 px-1 rounded">*</code> for every value,
|
||||
<code class="bg-gray-100 px-1 rounded">*/n</code> for every n-th value,
|
||||
<code class="bg-gray-100 px-1 rounded">0,6</code> for specific values.
|
||||
</p>
|
||||
<div class="grid grid-cols-2 gap-3 sm:grid-cols-5">
|
||||
<div>
|
||||
<label class="block text-xs font-medium text-gray-500 mb-1" for="cronMinute">Minute</label>
|
||||
<input id="cronMinute" type="text" x-model="editModal.form.cron_minute"
|
||||
class="w-full border border-gray-300 rounded-md px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-indigo-400"
|
||||
placeholder="0" required />
|
||||
</div>
|
||||
<div>
|
||||
<label class="block text-xs font-medium text-gray-500 mb-1" for="cronHour">Hour</label>
|
||||
<input id="cronHour" type="text" x-model="editModal.form.cron_hour"
|
||||
class="w-full border border-gray-300 rounded-md px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-indigo-400"
|
||||
placeholder="*" required />
|
||||
</div>
|
||||
<div>
|
||||
<label class="block text-xs font-medium text-gray-500 mb-1" for="cronDow">Day of Week</label>
|
||||
<input id="cronDow" type="text" x-model="editModal.form.cron_day_of_week"
|
||||
class="w-full border border-gray-300 rounded-md px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-indigo-400"
|
||||
placeholder="*" required />
|
||||
</div>
|
||||
<div>
|
||||
<label class="block text-xs font-medium text-gray-500 mb-1" for="cronDom">Day of Month</label>
|
||||
<input id="cronDom" type="text" x-model="editModal.form.cron_day_of_month"
|
||||
class="w-full border border-gray-300 rounded-md px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-indigo-400"
|
||||
placeholder="*" required />
|
||||
</div>
|
||||
<div>
|
||||
<label class="block text-xs font-medium text-gray-500 mb-1" for="cronMoy">Month</label>
|
||||
<input id="cronMoy" type="text" x-model="editModal.form.cron_month_of_year"
|
||||
class="w-full border border-gray-300 rounded-md px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-indigo-400"
|
||||
placeholder="*" required />
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Interval field -->
|
||||
<div x-show="editModal.form.schedule_type === 'interval'" class="mb-4">
|
||||
<label class="block text-sm font-medium text-gray-700 mb-1" for="intervalSeconds">Interval (seconds)</label>
|
||||
<input
|
||||
id="intervalSeconds"
|
||||
type="number"
|
||||
min="60"
|
||||
x-model.number="editModal.form.interval_seconds"
|
||||
class="w-full border border-gray-300 rounded-md px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-indigo-400"
|
||||
placeholder="3600"
|
||||
/>
|
||||
<p class="text-xs text-gray-500 mt-1">Minimum 60 seconds. Examples: 3600 = hourly, 86400 = daily.</p>
|
||||
</div>
|
||||
|
||||
<!-- Form errors -->
|
||||
<div x-show="editModal.error" class="mb-4 bg-red-50 border border-red-300 text-red-700 rounded p-3 text-sm" x-text="editModal.error" role="alert"></div>
|
||||
|
||||
<!-- Actions -->
|
||||
<div class="flex justify-end gap-3">
|
||||
<button
|
||||
type="button"
|
||||
@click="closeEditModal()"
|
||||
class="px-4 py-2 text-sm font-medium text-gray-700 bg-white border border-gray-300 rounded-md hover:bg-gray-50 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
|
||||
>
|
||||
Cancel
|
||||
</button>
|
||||
<button
|
||||
type="submit"
|
||||
:disabled="editModal.saving"
|
||||
class="px-4 py-2 text-sm font-medium text-white bg-indigo-600 hover:bg-indigo-700 disabled:opacity-50 disabled:cursor-not-allowed rounded-md focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
|
||||
>
|
||||
<i x-show="editModal.saving" class="fas fa-spinner fa-spin mr-2" aria-hidden="true"></i>
|
||||
Save Changes
|
||||
</button>
|
||||
</div>
|
||||
</form>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
{% endblock %}
|
||||
|
||||
{% block scripts %}
|
||||
<script>
|
||||
function scheduledJobsApp() {
|
||||
return {
|
||||
jobs: [],
|
||||
loading: true,
|
||||
runningJobIds: [],
|
||||
alert: { show: false, type: 'success', title: '', message: '' },
|
||||
editModal: {
|
||||
open: false,
|
||||
job: null,
|
||||
saving: false,
|
||||
error: null,
|
||||
form: {
|
||||
schedule_type: 'cron',
|
||||
cron_minute: '0',
|
||||
cron_hour: '*',
|
||||
cron_day_of_week: '*',
|
||||
cron_day_of_month: '*',
|
||||
cron_month_of_year: '*',
|
||||
interval_seconds: 3600,
|
||||
},
|
||||
},
|
||||
|
||||
init() {
|
||||
this.fetchJobs();
|
||||
},
|
||||
|
||||
async fetchJobs() {
|
||||
this.loading = true;
|
||||
try {
|
||||
const resp = await fetch('/api/admin/scheduled-jobs', { credentials: 'same-origin' });
|
||||
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
|
||||
this.jobs = await resp.json();
|
||||
} catch (err) {
|
||||
this.showAlert('error', 'Failed to load jobs', err.message);
|
||||
} finally {
|
||||
this.loading = false;
|
||||
}
|
||||
},
|
||||
|
||||
async toggleEnabled(job) {
|
||||
const newValue = !job.enabled;
|
||||
try {
|
||||
const resp = await fetch(`/api/admin/scheduled-jobs/${job.id}`, {
|
||||
method: 'PATCH',
|
||||
headers: this._headers(),
|
||||
body: JSON.stringify({ enabled: newValue }),
|
||||
credentials: 'same-origin',
|
||||
});
|
||||
if (!resp.ok) {
|
||||
const body = await resp.json().catch(() => ({}));
|
||||
throw new Error(body.detail || `HTTP ${resp.status}`);
|
||||
}
|
||||
const updated = await resp.json();
|
||||
const idx = this.jobs.findIndex(j => j.id === job.id);
|
||||
if (idx !== -1) this.jobs[idx] = updated;
|
||||
this.showAlert(
|
||||
'success',
|
||||
updated.enabled ? 'Job enabled' : 'Job disabled',
|
||||
`"${updated.display_name}" has been ${updated.enabled ? 'enabled' : 'disabled'}. Restart the worker for changes to take effect.`
|
||||
);
|
||||
} catch (err) {
|
||||
this.showAlert('error', 'Update failed', err.message);
|
||||
}
|
||||
},
|
||||
|
||||
async runNow(job) {
|
||||
if (this.runningJobIds.includes(job.id)) return;
|
||||
this.runningJobIds.push(job.id);
|
||||
try {
|
||||
const resp = await fetch(`/api/admin/scheduled-jobs/${job.id}/run-now`, {
|
||||
method: 'POST',
|
||||
headers: this._headers(),
|
||||
credentials: 'same-origin',
|
||||
});
|
||||
if (!resp.ok) {
|
||||
const body = await resp.json().catch(() => ({}));
|
||||
throw new Error(body.detail || `HTTP ${resp.status}`);
|
||||
}
|
||||
const data = await resp.json();
|
||||
this.showAlert(
|
||||
'success',
|
||||
'Job dispatched',
|
||||
`"${job.display_name}" has been queued (task ID: ${data.task_id}). Refresh to see the updated last-run status.`
|
||||
);
|
||||
// Refresh after a short delay so the status can update.
|
||||
setTimeout(() => this.fetchJobs(), 3000);
|
||||
} catch (err) {
|
||||
this.showAlert('error', 'Failed to dispatch job', err.message);
|
||||
} finally {
|
||||
this.runningJobIds = this.runningJobIds.filter(id => id !== job.id);
|
||||
}
|
||||
},
|
||||
|
||||
openEditModal(job) {
|
||||
this.editModal.job = job;
|
||||
this.editModal.error = null;
|
||||
this.editModal.saving = false;
|
||||
this.editModal.form = {
|
||||
schedule_type: job.schedule_type,
|
||||
cron_minute: job.cron_minute,
|
||||
cron_hour: job.cron_hour,
|
||||
cron_day_of_week: job.cron_day_of_week,
|
||||
cron_day_of_month: job.cron_day_of_month,
|
||||
cron_month_of_year: job.cron_month_of_year,
|
||||
interval_seconds: job.interval_seconds || 3600,
|
||||
};
|
||||
this.editModal.open = true;
|
||||
},
|
||||
|
||||
closeEditModal() {
|
||||
this.editModal.open = false;
|
||||
this.editModal.job = null;
|
||||
this.editModal.error = null;
|
||||
},
|
||||
|
||||
async saveSchedule() {
|
||||
if (!this.editModal.job) return;
|
||||
this.editModal.saving = true;
|
||||
this.editModal.error = null;
|
||||
|
||||
const payload = { schedule_type: this.editModal.form.schedule_type };
|
||||
if (this.editModal.form.schedule_type === 'cron') {
|
||||
payload.cron_minute = this.editModal.form.cron_minute;
|
||||
payload.cron_hour = this.editModal.form.cron_hour;
|
||||
payload.cron_day_of_week = this.editModal.form.cron_day_of_week;
|
||||
payload.cron_day_of_month = this.editModal.form.cron_day_of_month;
|
||||
payload.cron_month_of_year = this.editModal.form.cron_month_of_year;
|
||||
} else {
|
||||
const secs = parseInt(this.editModal.form.interval_seconds, 10);
|
||||
if (!secs || secs < 60) {
|
||||
this.editModal.error = 'Interval must be at least 60 seconds.';
|
||||
this.editModal.saving = false;
|
||||
return;
|
||||
}
|
||||
payload.interval_seconds = secs;
|
||||
}
|
||||
|
||||
try {
|
||||
const resp = await fetch(`/api/admin/scheduled-jobs/${this.editModal.job.id}`, {
|
||||
method: 'PATCH',
|
||||
headers: this._headers(),
|
||||
body: JSON.stringify(payload),
|
||||
credentials: 'same-origin',
|
||||
});
|
||||
if (!resp.ok) {
|
||||
const body = await resp.json().catch(() => ({}));
|
||||
throw new Error(body.detail || `HTTP ${resp.status}`);
|
||||
}
|
||||
const updated = await resp.json();
|
||||
const idx = this.jobs.findIndex(j => j.id === this.editModal.job.id);
|
||||
if (idx !== -1) this.jobs[idx] = updated;
|
||||
this.closeEditModal();
|
||||
this.showAlert(
|
||||
'success',
|
||||
'Schedule updated',
|
||||
`"${updated.display_name}" schedule saved. Restart the worker for changes to take effect.`
|
||||
);
|
||||
} catch (err) {
|
||||
this.editModal.error = err.message;
|
||||
} finally {
|
||||
this.editModal.saving = false;
|
||||
}
|
||||
},
|
||||
|
||||
formatSchedule(job) {
|
||||
if (job.schedule_type === 'interval') {
|
||||
const s = job.interval_seconds || 0;
|
||||
if (s >= 86400) return `Every ${s / 86400}d`;
|
||||
if (s >= 3600) return `Every ${s / 3600}h`;
|
||||
if (s >= 60) return `Every ${s / 60}m`;
|
||||
return `Every ${s}s`;
|
||||
}
|
||||
return `${job.cron_minute} ${job.cron_hour} ${job.cron_day_of_month} ${job.cron_month_of_year} ${job.cron_day_of_week}`;
|
||||
},
|
||||
|
||||
formatDate(isoStr) {
|
||||
try {
|
||||
return new Date(isoStr).toLocaleString(undefined, {
|
||||
dateStyle: 'short', timeStyle: 'short'
|
||||
});
|
||||
} catch {
|
||||
return isoStr;
|
||||
}
|
||||
},
|
||||
|
||||
showAlert(type, title, message) {
|
||||
this.alert = { show: true, type, title, message };
|
||||
setTimeout(() => { this.alert.show = false; }, 6000);
|
||||
},
|
||||
|
||||
_headers() {
|
||||
const meta = document.querySelector('meta[name="csrf-token"]');
|
||||
const headers = { 'Content-Type': 'application/json' };
|
||||
if (meta) headers['X-CSRF-Token'] = meta.content;
|
||||
return headers;
|
||||
},
|
||||
};
|
||||
}
|
||||
</script>
|
||||
{% endblock %}
|
||||
@@ -165,6 +165,9 @@
|
||||
<a href="/admin/queue" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
|
||||
<i class="fas fa-stream w-4 mr-2 text-blue-500" aria-hidden="true"></i> Queue Monitor
|
||||
</a>
|
||||
<a href="/admin/scheduled-jobs" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
|
||||
<i class="fas fa-clock w-4 mr-2 text-indigo-500" aria-hidden="true"></i> Scheduled Jobs
|
||||
</a>
|
||||
<a href="/admin/backup" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
|
||||
<i class="fas fa-database w-4 mr-2 text-green-600" aria-hidden="true"></i> Backup & Restore
|
||||
</a>
|
||||
@@ -332,6 +335,9 @@
|
||||
<a href="/admin/queue" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
|
||||
<i class="fas fa-stream mr-2 text-blue-400" aria-hidden="true"></i> Queue Monitor
|
||||
</a>
|
||||
<a href="/admin/scheduled-jobs" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
|
||||
<i class="fas fa-clock mr-2 text-indigo-500" aria-hidden="true"></i> Scheduled Jobs
|
||||
</a>
|
||||
<a href="/admin/backup" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
|
||||
<i class="fas fa-database mr-2 text-green-500" aria-hidden="true"></i> Backup & Restore
|
||||
</a>
|
||||
|
||||
@@ -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")
|
||||
@@ -67,6 +67,7 @@ from app.models import ( # noqa: F401, E402
|
||||
PipelineStep,
|
||||
ProcessingLog,
|
||||
SavedSearch,
|
||||
ScheduledJob,
|
||||
UserImapAccount,
|
||||
UserIntegration,
|
||||
UserProfile,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user