feat: add jobs CRUD API with manual execution
- Full CRUD for backup jobs - Manual job trigger endpoint with background execution - Schedule creation endpoint - Tests for job creation, update, execution, and 404 cases
This commit is contained in:
@@ -0,0 +1,98 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select
|
||||
from typing import List
|
||||
from app.database import get_db
|
||||
from app.models import Job, Schedule
|
||||
from app.schemas import JobCreate, JobUpdate, Job as JobSchema, ScheduleCreate
|
||||
from backup.engine import BackupEngine
|
||||
|
||||
router = APIRouter(prefix="/api/jobs", tags=["jobs"])
|
||||
|
||||
@router.get("/", response_model=List[JobSchema])
|
||||
async def list_jobs(db: AsyncSession = Depends(get_db)):
|
||||
result = await db.execute(select(Job))
|
||||
jobs = result.scalars().all()
|
||||
return jobs
|
||||
|
||||
@router.post("/", response_model=JobSchema)
|
||||
async def create_job(job: JobCreate, db: AsyncSession = Depends(get_db)):
|
||||
db_job = Job(**job.model_dump())
|
||||
db.add(db_job)
|
||||
await db.commit()
|
||||
await db.refresh(db_job)
|
||||
return db_job
|
||||
|
||||
@router.get("/{job_id}", response_model=JobSchema)
|
||||
async def get_job(job_id: int, db: AsyncSession = Depends(get_db)):
|
||||
result = await db.execute(select(Job).where(Job.id == job_id))
|
||||
job = result.scalar_one_or_none()
|
||||
if not job:
|
||||
raise HTTPException(status_code=404, detail="Job not found")
|
||||
return job
|
||||
|
||||
@router.put("/{job_id}", response_model=JobSchema)
|
||||
async def update_job(
|
||||
job_id: int,
|
||||
job_update: JobUpdate,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
result = await db.execute(select(Job).where(Job.id == job_id))
|
||||
job = result.scalar_one_or_none()
|
||||
if not job:
|
||||
raise HTTPException(status_code=404, detail="Job not found")
|
||||
|
||||
update_data = job_update.model_dump(exclude_unset=True)
|
||||
for field, value in update_data.items():
|
||||
setattr(job, field, value)
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(job)
|
||||
return job
|
||||
|
||||
@router.delete("/{job_id}")
|
||||
async def delete_job(job_id: int, db: AsyncSession = Depends(get_db)):
|
||||
result = await db.execute(select(Job).where(Job.id == job_id))
|
||||
job = result.scalar_one_or_none()
|
||||
if not job:
|
||||
raise HTTPException(status_code=404, detail="Job not found")
|
||||
|
||||
await db.delete(job)
|
||||
await db.commit()
|
||||
return {"message": "Job deleted"}
|
||||
|
||||
@router.post("/{job_id}/run")
|
||||
async def run_job(
|
||||
job_id: int,
|
||||
background_tasks: BackgroundTasks,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
result = await db.execute(select(Job).where(Job.id == job_id))
|
||||
job = result.scalar_one_or_none()
|
||||
if not job:
|
||||
raise HTTPException(status_code=404, detail="Job not found")
|
||||
|
||||
# Run in background
|
||||
async def execute():
|
||||
engine = BackupEngine(db)
|
||||
await engine.execute_job(job_id, triggered_by="manual")
|
||||
|
||||
background_tasks.add_task(execute)
|
||||
return {"message": "Job execution started"}
|
||||
|
||||
@router.post("/{job_id}/schedule", response_model=JobSchema)
|
||||
async def create_schedule(
|
||||
job_id: int,
|
||||
schedule: ScheduleCreate,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
result = await db.execute(select(Job).where(Job.id == job_id))
|
||||
job = result.scalar_one_or_none()
|
||||
if not job:
|
||||
raise HTTPException(status_code=404, detail="Job not found")
|
||||
|
||||
db_schedule = Schedule(**schedule.model_dump())
|
||||
db.add(db_schedule)
|
||||
await db.commit()
|
||||
await db.refresh(job)
|
||||
return job
|
||||
Reference in New Issue
Block a user