alex
955ed9db49
- Use new session in background task for thread safety - Fix schedule creation to associate with job - Fix schedule endpoint return type - Update tests to use AsyncClient - Add missing tests for list, get, delete, schedule - Add updated_at field to Schedule model
102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
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, AsyncSessionLocal
|
|
from app.models import Job, Schedule
|
|
from app.schemas import JobCreate, JobUpdate, Job as JobSchema, ScheduleCreate, Schedule as ScheduleSchema
|
|
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():
|
|
async with AsyncSessionLocal() as session:
|
|
engine = BackupEngine(session)
|
|
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=ScheduleSchema)
|
|
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")
|
|
|
|
schedule_data = schedule.model_dump()
|
|
schedule_data["job_id"] = job_id
|
|
db_schedule = Schedule(**schedule_data)
|
|
db.add(db_schedule)
|
|
await db.commit()
|
|
await db.refresh(db_schedule)
|
|
return ScheduleSchema.model_validate(db_schedule)
|