feat(fc2a): add /api/import admin endpoints
trigger enqueues a Celery scan_directory task and returns 202 with the celery task id (the actual scan happens asynchronously; status surfaces via /api/import/status and /api/system/stats). list_tasks supports status filter and cursor pagination. retry-failed resets failed tasks to queued and re-enqueues. clear-completed deletes finished tasks older than the supplied age (used by the Settings UI's cleanup control). mode='deep' is explicitly rejected with 400 until FC-2d ships it; the frontend's UI in FC-2a only sends 'quick'. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -15,6 +15,7 @@ api_bp.add_url_rule("/health", view_func=health.get_health, methods=["GET"])
|
||||
|
||||
def all_blueprints() -> list[Blueprint]:
|
||||
from .gallery import gallery_bp
|
||||
from .import_admin import import_admin_bp
|
||||
from .settings import settings_bp
|
||||
from .tags import tags_bp
|
||||
return [api_bp, gallery_bp, tags_bp, settings_bp]
|
||||
return [api_bp, gallery_bp, tags_bp, settings_bp, import_admin_bp]
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Import admin API: trigger scan, list tasks, retry, clear."""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from quart import Blueprint, jsonify, request
|
||||
from sqlalchemy import delete, select, update
|
||||
|
||||
from ..extensions import make_engine, make_session_factory
|
||||
from ..models import ImportBatch, ImportTask
|
||||
|
||||
import_admin_bp = Blueprint("import_admin", __name__, url_prefix="/api/import")
|
||||
|
||||
_engine = None
|
||||
_Session = None
|
||||
|
||||
|
||||
def _session_factory():
|
||||
global _engine, _Session
|
||||
if _engine is None:
|
||||
_engine = make_engine()
|
||||
_Session = make_session_factory(_engine)
|
||||
return _Session
|
||||
|
||||
|
||||
@import_admin_bp.route("/trigger", methods=["POST"])
|
||||
async def trigger_scan():
|
||||
body = await request.get_json(silent=True) or {}
|
||||
mode = body.get("mode", "quick")
|
||||
if mode != "quick":
|
||||
# FC-2d will accept 'deep'. For now, reject anything else.
|
||||
return jsonify({"error": f"mode {mode!r} not supported in FC-2a; use 'quick'"}), 400
|
||||
|
||||
# Enqueue the scan task in Celery. The task creates the batch itself.
|
||||
from ..tasks.scan import scan_directory
|
||||
|
||||
async_result = scan_directory.delay(triggered_by="manual")
|
||||
return jsonify({"celery_task_id": async_result.id, "mode": mode}), 202
|
||||
|
||||
|
||||
@import_admin_bp.route("/status", methods=["GET"])
|
||||
async def status():
|
||||
Session = _session_factory()
|
||||
async with Session() as session:
|
||||
active = (
|
||||
await session.execute(
|
||||
select(ImportBatch)
|
||||
.where(ImportBatch.status == "running")
|
||||
.order_by(ImportBatch.started_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
payload = {"active_batch": None}
|
||||
if active:
|
||||
payload["active_batch"] = {
|
||||
"id": active.id,
|
||||
"total_files": active.total_files,
|
||||
"imported": active.imported,
|
||||
"skipped": active.skipped,
|
||||
"failed": active.failed,
|
||||
"started_at": active.started_at.isoformat(),
|
||||
}
|
||||
return jsonify(payload)
|
||||
|
||||
|
||||
@import_admin_bp.route("/tasks", methods=["GET"])
|
||||
async def list_tasks():
|
||||
status_filter = request.args.get("status")
|
||||
try:
|
||||
limit = min(int(request.args.get("limit", "50")), 200)
|
||||
except ValueError:
|
||||
return jsonify({"error": "limit must be an integer"}), 400
|
||||
cursor_raw = request.args.get("cursor")
|
||||
cursor_id = int(cursor_raw) if cursor_raw else None
|
||||
|
||||
Session = _session_factory()
|
||||
async with Session() as session:
|
||||
stmt = select(ImportTask).order_by(ImportTask.created_at.desc(), ImportTask.id.desc())
|
||||
if status_filter:
|
||||
stmt = stmt.where(ImportTask.status == status_filter)
|
||||
if cursor_id is not None:
|
||||
stmt = stmt.where(ImportTask.id < cursor_id)
|
||||
stmt = stmt.limit(limit + 1)
|
||||
rows = (await session.execute(stmt)).scalars().all()
|
||||
|
||||
has_more = len(rows) > limit
|
||||
rows = rows[:limit]
|
||||
return jsonify({
|
||||
"tasks": [
|
||||
{
|
||||
"id": t.id,
|
||||
"batch_id": t.batch_id,
|
||||
"source_path": t.source_path,
|
||||
"task_type": t.task_type,
|
||||
"status": t.status,
|
||||
"result_image_id": t.result_image_id,
|
||||
"error": t.error,
|
||||
"size_bytes": t.size_bytes,
|
||||
"created_at": t.created_at.isoformat(),
|
||||
"started_at": t.started_at.isoformat() if t.started_at else None,
|
||||
"finished_at": t.finished_at.isoformat() if t.finished_at else None,
|
||||
}
|
||||
for t in rows
|
||||
],
|
||||
"next_cursor": rows[-1].id if has_more and rows else None,
|
||||
})
|
||||
|
||||
|
||||
@import_admin_bp.route("/retry-failed", methods=["POST"])
|
||||
async def retry_failed():
|
||||
Session = _session_factory()
|
||||
async with Session() as session:
|
||||
failed_ids = (
|
||||
await session.execute(select(ImportTask.id).where(ImportTask.status == "failed"))
|
||||
).scalars().all()
|
||||
if not failed_ids:
|
||||
return jsonify({"retried": 0})
|
||||
await session.execute(
|
||||
update(ImportTask)
|
||||
.where(ImportTask.id.in_(failed_ids))
|
||||
.values(status="queued", error=None, started_at=None, finished_at=None)
|
||||
)
|
||||
await session.commit()
|
||||
|
||||
from ..tasks.import_file import import_media_file
|
||||
for tid in failed_ids:
|
||||
import_media_file.delay(tid)
|
||||
|
||||
return jsonify({"retried": len(failed_ids)})
|
||||
|
||||
|
||||
@import_admin_bp.route("/clear-completed", methods=["POST"])
|
||||
async def clear_completed():
|
||||
body = await request.get_json(silent=True) or {}
|
||||
age_days = body.get("age_days", 0)
|
||||
status_filter = body.get("status", ["complete", "skipped"])
|
||||
if not isinstance(status_filter, list):
|
||||
return jsonify({"error": "status must be a list"}), 400
|
||||
|
||||
cutoff = (
|
||||
datetime.now(timezone.utc) - timedelta(days=int(age_days)) if age_days else None
|
||||
)
|
||||
|
||||
Session = _session_factory()
|
||||
async with Session() as session:
|
||||
stmt = delete(ImportTask).where(ImportTask.status.in_(status_filter))
|
||||
if cutoff is not None:
|
||||
stmt = stmt.where(ImportTask.finished_at < cutoff)
|
||||
result = await session.execute(stmt)
|
||||
await session.commit()
|
||||
|
||||
return jsonify({"deleted": result.rowcount or 0})
|
||||
@@ -0,0 +1,90 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app import create_app
|
||||
from backend.app.celery_app import celery
|
||||
from backend.app.models import ImportBatch, ImportTask
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def eager():
|
||||
celery.conf.task_always_eager = True
|
||||
yield
|
||||
celery.conf.task_always_eager = False
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def app():
|
||||
return create_app()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client(app):
|
||||
async with app.test_client() as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_rejects_unknown_mode(client):
|
||||
resp = await client.post("/api/import/trigger", json={"mode": "deep"})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_status_when_idle(client):
|
||||
resp = await client.get("/api/import/status")
|
||||
body = await resp.get_json()
|
||||
assert body["active_batch"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_tasks_with_status_filter(client, db):
|
||||
batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
|
||||
db.add(batch)
|
||||
await db.flush()
|
||||
for status in ("complete", "failed", "complete"):
|
||||
db.add(ImportTask(
|
||||
batch_id=batch.id, source_path="/x", task_type="media", status=status,
|
||||
finished_at=datetime.now(timezone.utc),
|
||||
))
|
||||
await db.commit()
|
||||
|
||||
resp = await client.get("/api/import/tasks?status=failed")
|
||||
body = await resp.get_json()
|
||||
assert len(body["tasks"]) == 1
|
||||
assert body["tasks"][0]["status"] == "failed"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_retry_failed(client, db):
|
||||
batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
|
||||
db.add(batch)
|
||||
await db.flush()
|
||||
db.add(ImportTask(batch_id=batch.id, source_path="/x", task_type="media", status="failed"))
|
||||
await db.commit()
|
||||
|
||||
resp = await client.post("/api/import/retry-failed")
|
||||
body = await resp.get_json()
|
||||
assert body["retried"] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clear_completed(client, db):
|
||||
batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
|
||||
db.add(batch)
|
||||
await db.flush()
|
||||
long_ago = datetime.now(timezone.utc) - timedelta(days=30)
|
||||
db.add(ImportTask(
|
||||
batch_id=batch.id, source_path="/x", task_type="media",
|
||||
status="complete", finished_at=long_ago,
|
||||
))
|
||||
db.add(ImportTask(
|
||||
batch_id=batch.id, source_path="/y", task_type="media",
|
||||
status="complete", finished_at=datetime.now(timezone.utc),
|
||||
))
|
||||
await db.commit()
|
||||
|
||||
resp = await client.post("/api/import/clear-completed", json={"age_days": 7})
|
||||
body = await resp.get_json()
|
||||
assert body["deleted"] == 1
|
||||
Reference in New Issue
Block a user