252 lines
7.9 KiB
Python
252 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import sqlalchemy as sa
|
|
from fastapi import APIRouter, Depends, Request
|
|
|
|
from app.api.admin.deps import get_admin_claims
|
|
from app.core.errors import ApiError, ErrorCode
|
|
from app.core.request_id import get_or_create_request_id
|
|
from app.core.time import iso_utc_ms
|
|
from app.db.models.jobs import JobRow
|
|
from app.db.session import create_sessionmaker, with_sqlite_busy_retry
|
|
|
|
router = APIRouter()
|
|
|
|
_ALLOWED_JOB_STATUSES = {"pending", "running", "paused", "canceled", "completed", "failed", "dlq"}
|
|
|
|
|
|
@router.get("/jobs")
|
|
async def list_jobs(
|
|
request: Request,
|
|
limit: int = 50,
|
|
cursor: str | None = None,
|
|
status: str | None = None,
|
|
type: str | None = None,
|
|
_claims: dict[str, Any] = Depends(get_admin_claims),
|
|
) -> dict[str, Any]:
|
|
_ = _claims
|
|
if limit < 1 or limit > 200:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Unsupported limit", status_code=400)
|
|
|
|
cursor_i: int | None = None
|
|
cursor_raw = (cursor or "").strip()
|
|
if cursor_raw:
|
|
if not cursor_raw.isdigit():
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Unsupported cursor", status_code=400)
|
|
cursor_i = int(cursor_raw)
|
|
if cursor_i <= 0:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Unsupported cursor", status_code=400)
|
|
|
|
status_norm = (status or "").strip().lower() or None
|
|
if status_norm is not None and status_norm not in _ALLOWED_JOB_STATUSES:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Unsupported status", status_code=400)
|
|
|
|
type_norm = (type or "").strip() or None
|
|
if type_norm is not None and len(type_norm) > 100:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Unsupported type", status_code=400)
|
|
|
|
rid = get_or_create_request_id(request)
|
|
|
|
engine = request.app.state.engine
|
|
Session = create_sessionmaker(engine)
|
|
|
|
stmt = sa.select(JobRow).order_by(JobRow.id.desc()).limit(limit + 1)
|
|
if cursor_i is not None:
|
|
stmt = stmt.where(JobRow.id < cursor_i)
|
|
if status_norm is not None:
|
|
stmt = stmt.where(JobRow.status == status_norm)
|
|
if type_norm is not None:
|
|
stmt = stmt.where(JobRow.type == type_norm)
|
|
|
|
async with Session() as session:
|
|
rows = ((await session.execute(stmt)).scalars().all())
|
|
|
|
items_rows = rows[:limit]
|
|
next_cursor = int(items_rows[-1].id) if len(rows) > limit and items_rows else None
|
|
|
|
items = [
|
|
{
|
|
"id": str(row.id),
|
|
"type": row.type,
|
|
"status": row.status,
|
|
"priority": int(row.priority),
|
|
"run_after": row.run_after,
|
|
"attempt": int(row.attempt),
|
|
"max_attempts": int(row.max_attempts),
|
|
"last_error": row.last_error,
|
|
"locked_by": row.locked_by,
|
|
"locked_at": row.locked_at,
|
|
"ref_type": row.ref_type,
|
|
"ref_id": row.ref_id,
|
|
"created_at": row.created_at,
|
|
"updated_at": row.updated_at,
|
|
}
|
|
for row in items_rows
|
|
]
|
|
|
|
return {
|
|
"ok": True,
|
|
"items": items,
|
|
"next_cursor": str(next_cursor) if next_cursor is not None else "",
|
|
"request_id": rid,
|
|
}
|
|
|
|
|
|
@router.get("/jobs/{job_id}")
|
|
async def get_job(
|
|
job_id: int,
|
|
request: Request,
|
|
_claims: dict[str, Any] = Depends(get_admin_claims),
|
|
) -> dict[str, Any]:
|
|
_ = _claims
|
|
if job_id <= 0:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Invalid job id", status_code=400)
|
|
|
|
rid = get_or_create_request_id(request)
|
|
|
|
engine = request.app.state.engine
|
|
Session = create_sessionmaker(engine)
|
|
|
|
async with Session() as session:
|
|
row = await session.get(JobRow, job_id)
|
|
if row is None:
|
|
raise ApiError(code=ErrorCode.NOT_FOUND, message="Job not found", status_code=404)
|
|
|
|
payload_json = str(row.payload_json or "")
|
|
payload: Any = None
|
|
if payload_json.strip():
|
|
try:
|
|
payload = json.loads(payload_json)
|
|
except Exception:
|
|
payload = None
|
|
|
|
return {
|
|
"ok": True,
|
|
"item": {
|
|
"id": str(row.id),
|
|
"type": row.type,
|
|
"status": row.status,
|
|
"priority": int(row.priority),
|
|
"run_after": row.run_after,
|
|
"attempt": int(row.attempt),
|
|
"max_attempts": int(row.max_attempts),
|
|
"payload": payload,
|
|
"payload_json": payload_json,
|
|
"last_error": row.last_error,
|
|
"locked_by": row.locked_by,
|
|
"locked_at": row.locked_at,
|
|
"ref_type": row.ref_type,
|
|
"ref_id": row.ref_id,
|
|
"created_at": row.created_at,
|
|
"updated_at": row.updated_at,
|
|
},
|
|
"request_id": rid,
|
|
}
|
|
|
|
|
|
@router.post("/jobs/{job_id}/retry")
|
|
async def retry_job(
|
|
job_id: int,
|
|
request: Request,
|
|
_claims: dict[str, Any] = Depends(get_admin_claims),
|
|
) -> dict[str, Any]:
|
|
_ = _claims
|
|
if job_id <= 0:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Invalid job id", status_code=400)
|
|
|
|
rid = get_or_create_request_id(request)
|
|
now = iso_utc_ms()
|
|
|
|
engine = request.app.state.engine
|
|
Session = create_sessionmaker(engine)
|
|
|
|
async def _op() -> dict[str, Any]:
|
|
async with Session() as session:
|
|
row = await session.get(JobRow, job_id)
|
|
if row is None:
|
|
raise ApiError(code=ErrorCode.NOT_FOUND, message="Job not found", status_code=404)
|
|
|
|
if row.status == "running":
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Job is running", status_code=400)
|
|
|
|
row.status = "pending"
|
|
row.run_after = None
|
|
row.locked_by = None
|
|
row.locked_at = None
|
|
row.updated_at = now
|
|
await session.commit()
|
|
|
|
return {"ok": True, "job_id": str(job_id), "status": "pending", "request_id": rid}
|
|
|
|
return await with_sqlite_busy_retry(_op)
|
|
|
|
|
|
@router.post("/jobs/{job_id}/cancel")
|
|
async def cancel_job(
|
|
job_id: int,
|
|
request: Request,
|
|
_claims: dict[str, Any] = Depends(get_admin_claims),
|
|
) -> dict[str, Any]:
|
|
_ = _claims
|
|
if job_id <= 0:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Invalid job id", status_code=400)
|
|
|
|
rid = get_or_create_request_id(request)
|
|
now = iso_utc_ms()
|
|
|
|
engine = request.app.state.engine
|
|
Session = create_sessionmaker(engine)
|
|
|
|
async def _op() -> dict[str, Any]:
|
|
async with Session() as session:
|
|
row = await session.get(JobRow, job_id)
|
|
if row is None:
|
|
raise ApiError(code=ErrorCode.NOT_FOUND, message="Job not found", status_code=404)
|
|
|
|
row.status = "canceled"
|
|
row.locked_by = None
|
|
row.locked_at = None
|
|
row.updated_at = now
|
|
await session.commit()
|
|
|
|
return {"ok": True, "job_id": str(job_id), "status": "canceled", "request_id": rid}
|
|
|
|
return await with_sqlite_busy_retry(_op)
|
|
|
|
|
|
@router.post("/jobs/{job_id}/move-to-dlq")
|
|
async def move_job_to_dlq(
|
|
job_id: int,
|
|
request: Request,
|
|
_claims: dict[str, Any] = Depends(get_admin_claims),
|
|
) -> dict[str, Any]:
|
|
_ = _claims
|
|
if job_id <= 0:
|
|
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Invalid job id", status_code=400)
|
|
|
|
rid = get_or_create_request_id(request)
|
|
now = iso_utc_ms()
|
|
|
|
engine = request.app.state.engine
|
|
Session = create_sessionmaker(engine)
|
|
|
|
async def _op() -> dict[str, Any]:
|
|
async with Session() as session:
|
|
row = await session.get(JobRow, job_id)
|
|
if row is None:
|
|
raise ApiError(code=ErrorCode.NOT_FOUND, message="Job not found", status_code=404)
|
|
|
|
row.status = "dlq"
|
|
row.run_after = None
|
|
row.locked_by = None
|
|
row.locked_at = None
|
|
row.updated_at = now
|
|
await session.commit()
|
|
|
|
return {"ok": True, "job_id": str(job_id), "status": "dlq", "request_id": rid}
|
|
|
|
return await with_sqlite_busy_retry(_op)
|