feat(fc2a): add /api/settings/import and /api/system/stats endpoints
Settings GET returns the single-row ImportSettings DB record; PATCH does selective field updates (only known editable fields are applied; unknown keys silently ignored). System stats returns image/tag/storage counts, task status histogram, and the active batch summary in one shot for the Settings overview tab. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,123 @@
|
||||
"""Settings API: import filters, system stats."""
|
||||
|
||||
from quart import Blueprint, jsonify, request
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from ..extensions import make_engine, make_session_factory
|
||||
from ..models import ImageRecord, ImportBatch, ImportSettings, ImportTask, Tag
|
||||
|
||||
settings_bp = Blueprint("settings", __name__, url_prefix="/api")
|
||||
|
||||
_engine = None
|
||||
_Session = None
|
||||
|
||||
|
||||
def _session_factory():
|
||||
global _engine, _Session
|
||||
if _engine is None:
|
||||
_engine = make_engine()
|
||||
_Session = make_session_factory(_engine)
|
||||
return _Session
|
||||
|
||||
|
||||
_EDITABLE_FIELDS = (
|
||||
"min_width",
|
||||
"min_height",
|
||||
"skip_transparent",
|
||||
"transparency_threshold",
|
||||
"skip_single_color",
|
||||
"single_color_threshold",
|
||||
"single_color_tolerance",
|
||||
)
|
||||
|
||||
|
||||
@settings_bp.route("/settings/import", methods=["GET"])
|
||||
async def get_import_settings():
|
||||
Session = _session_factory()
|
||||
async with Session() as session:
|
||||
row = (
|
||||
await session.execute(select(ImportSettings).where(ImportSettings.id == 1))
|
||||
).scalar_one()
|
||||
return jsonify({
|
||||
"min_width": row.min_width,
|
||||
"min_height": row.min_height,
|
||||
"skip_transparent": row.skip_transparent,
|
||||
"transparency_threshold": row.transparency_threshold,
|
||||
"skip_single_color": row.skip_single_color,
|
||||
"single_color_threshold": row.single_color_threshold,
|
||||
"single_color_tolerance": row.single_color_tolerance,
|
||||
})
|
||||
|
||||
|
||||
@settings_bp.route("/settings/import", methods=["PATCH"])
|
||||
async def update_import_settings():
|
||||
body = await request.get_json()
|
||||
if not isinstance(body, dict):
|
||||
return jsonify({"error": "body must be a JSON object"}), 400
|
||||
|
||||
Session = _session_factory()
|
||||
async with Session() as session:
|
||||
row = (
|
||||
await session.execute(select(ImportSettings).where(ImportSettings.id == 1))
|
||||
).scalar_one()
|
||||
for field in _EDITABLE_FIELDS:
|
||||
if field in body:
|
||||
setattr(row, field, body[field])
|
||||
await session.commit()
|
||||
|
||||
return await get_import_settings()
|
||||
|
||||
|
||||
@settings_bp.route("/system/stats", methods=["GET"])
|
||||
async def system_stats():
|
||||
Session = _session_factory()
|
||||
async with Session() as session:
|
||||
total_images = (await session.execute(select(func.count(ImageRecord.id)))).scalar_one()
|
||||
total_tags = (await session.execute(select(func.count(Tag.id)))).scalar_one()
|
||||
storage_bytes = (
|
||||
(await session.execute(select(func.coalesce(func.sum(ImageRecord.size_bytes), 0)))).scalar_one()
|
||||
)
|
||||
|
||||
# Task counts grouped by status
|
||||
status_rows = (
|
||||
await session.execute(
|
||||
select(ImportTask.status, func.count(ImportTask.id)).group_by(ImportTask.status)
|
||||
)
|
||||
).all()
|
||||
status_counts = {row[0]: row[1] for row in status_rows}
|
||||
|
||||
# Active batch (most recent running)
|
||||
active_batch_row = (
|
||||
await session.execute(
|
||||
select(ImportBatch)
|
||||
.where(ImportBatch.status == "running")
|
||||
.order_by(ImportBatch.started_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
active_batch = None
|
||||
if active_batch_row:
|
||||
active_batch = {
|
||||
"id": active_batch_row.id,
|
||||
"source_path": active_batch_row.source_path,
|
||||
"started_at": active_batch_row.started_at.isoformat(),
|
||||
"total_files": active_batch_row.total_files,
|
||||
"imported": active_batch_row.imported,
|
||||
"skipped": active_batch_row.skipped,
|
||||
"failed": active_batch_row.failed,
|
||||
}
|
||||
|
||||
return jsonify({
|
||||
"total_images": total_images,
|
||||
"total_tags": total_tags,
|
||||
"storage_bytes": storage_bytes,
|
||||
"tasks": {
|
||||
"pending": status_counts.get("pending", 0),
|
||||
"queued": status_counts.get("queued", 0),
|
||||
"processing": status_counts.get("processing", 0),
|
||||
"complete": status_counts.get("complete", 0),
|
||||
"skipped": status_counts.get("skipped", 0),
|
||||
"failed": status_counts.get("failed", 0),
|
||||
},
|
||||
"active_batch": active_batch,
|
||||
})
|
||||
@@ -0,0 +1,56 @@
|
||||
import pytest
|
||||
|
||||
from backend.app import create_app
|
||||
|
||||
|
||||
@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_get_import_settings_defaults(client):
|
||||
resp = await client.get("/api/settings/import")
|
||||
assert resp.status_code == 200
|
||||
body = await resp.get_json()
|
||||
# Defaults from the migration:
|
||||
assert body["min_width"] == 0
|
||||
assert body["min_height"] == 0
|
||||
assert body["skip_transparent"] is False
|
||||
assert body["transparency_threshold"] == 0.9
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_patch_import_settings(client):
|
||||
resp = await client.patch(
|
||||
"/api/settings/import",
|
||||
json={"min_width": 500, "skip_transparent": True, "transparency_threshold": 0.7},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = await resp.get_json()
|
||||
assert body["min_width"] == 500
|
||||
assert body["skip_transparent"] is True
|
||||
assert body["transparency_threshold"] == 0.7
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_patch_rejects_non_object(client):
|
||||
resp = await client.patch("/api/settings/import", json=[1, 2, 3])
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_system_stats_shape(client):
|
||||
resp = await client.get("/api/system/stats")
|
||||
assert resp.status_code == 200
|
||||
body = await resp.get_json()
|
||||
for key in ("total_images", "total_tags", "storage_bytes", "tasks", "active_batch"):
|
||||
assert key in body
|
||||
for status in ("pending", "queued", "processing", "complete", "skipped", "failed"):
|
||||
assert status in body["tasks"]
|
||||
Reference in New Issue
Block a user