diff --git a/backend/app/main.py b/backend/app/main.py index 2f0753c..dc72052 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -5,12 +5,16 @@ from contextlib import asynccontextmanager from app.database import engine, Base from app import models # noqa: F401 - registers models with Base.metadata from app.routers import sources, jobs, executions, backups, settings, dashboard +from backup.scheduler import backup_scheduler @asynccontextmanager async def lifespan(app: FastAPI): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) + backup_scheduler.start() + await backup_scheduler.sync_schedules() yield + backup_scheduler.shutdown() app = FastAPI( title="Backup Tool API", diff --git a/backend/backup/scheduler.py b/backend/backup/scheduler.py new file mode 100644 index 0000000..43d7924 --- /dev/null +++ b/backend/backup/scheduler.py @@ -0,0 +1,81 @@ +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger +from sqlalchemy import select +from typing import Optional +import logging + +from app.database import AsyncSessionLocal +from app.models import Schedule +from backup.engine import BackupEngine + +logger = logging.getLogger(__name__) + + +class BackupScheduler: + def __init__(self): + self.scheduler = AsyncIOScheduler() + self._job_map = {} + + def start(self): + """Start the scheduler.""" + self.scheduler.start() + logger.info("Backup scheduler started") + + def shutdown(self): + """Shutdown the scheduler.""" + self.scheduler.shutdown() + logger.info("Backup scheduler shutdown") + + async def sync_schedules(self): + """Sync all enabled schedules from database.""" + async with AsyncSessionLocal() as db: + result = await db.execute( + select(Schedule).where(Schedule.enabled == True) + ) + schedules = result.scalars().all() + + # Clear existing jobs + for schedule_id, job_id in list(self._job_map.items()): + self.scheduler.remove_job(job_id) + del self._job_map[schedule_id] + + # Add new jobs + for schedule in schedules: + await self._add_schedule_job(schedule) + + async def _add_schedule_job(self, schedule: Schedule): + """Add a single schedule job to the scheduler.""" + try: + trigger = CronTrigger.from_crontab(schedule.cron_expression) + job = self.scheduler.add_job( + self._run_backup_job, + trigger=trigger, + args=[schedule.job_id], + id=f"backup_job_{schedule.job_id}", + replace_existing=True + ) + self._job_map[schedule.id] = job.id + logger.info(f"Scheduled backup job {schedule.job_id} with cron: {schedule.cron_expression}") + except Exception as e: + logger.error(f"Failed to schedule job {schedule.job_id}: {e}") + + async def _run_backup_job(self, job_id: int): + """Execute a backup job.""" + logger.info(f"Running scheduled backup job {job_id}") + async with AsyncSessionLocal() as db: + engine = BackupEngine(db) + await engine.execute_job(job_id, triggered_by="schedule") + + async def add_schedule(self, schedule: Schedule): + """Add a new schedule to the scheduler.""" + await self._add_schedule_job(schedule) + + def remove_schedule(self, schedule_id: int): + """Remove a schedule from the scheduler.""" + if schedule_id in self._job_map: + self.scheduler.remove_job(self._job_map[schedule_id]) + del self._job_map[schedule_id] + + +# Global scheduler instance +backup_scheduler = BackupScheduler()