122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
"""FC-3a: CRUD over Source rows."""
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
|
|
from ..extensions import get_session
|
|
from ..services.source_service import (
|
|
KNOWN_PLATFORMS,
|
|
ArtistNotFoundError,
|
|
DuplicateSourceError,
|
|
EmptyUrlError,
|
|
InvalidConfigError,
|
|
SourceService,
|
|
UnknownPlatformError,
|
|
)
|
|
|
|
sources_bp = Blueprint("sources", __name__, url_prefix="/api/sources")
|
|
|
|
|
|
def _bad(error: str, *, status: int = 400, detail: str | None = None, **extra):
|
|
body = {"error": error}
|
|
if detail is not None:
|
|
body["detail"] = detail
|
|
body.update(extra)
|
|
return jsonify(body), status
|
|
|
|
|
|
@sources_bp.route("/platforms", methods=["GET"])
|
|
async def platforms():
|
|
return jsonify({"platforms": sorted(KNOWN_PLATFORMS)})
|
|
|
|
|
|
@sources_bp.route("", methods=["GET"])
|
|
async def list_sources():
|
|
artist_id_raw = request.args.get("artist_id")
|
|
artist_id = None
|
|
if artist_id_raw is not None:
|
|
try:
|
|
artist_id = int(artist_id_raw)
|
|
except ValueError:
|
|
return _bad("invalid_artist_id", detail="artist_id must be an integer")
|
|
async with get_session() as session:
|
|
records = await SourceService(session).list(artist_id=artist_id)
|
|
return jsonify([r.to_dict() for r in records])
|
|
|
|
|
|
@sources_bp.route("/<int:source_id>", methods=["GET"])
|
|
async def get_source(source_id: int):
|
|
async with get_session() as session:
|
|
record = await SourceService(session).get(source_id)
|
|
if record is None:
|
|
return _bad("not_found", status=404, detail=f"source id={source_id}")
|
|
return jsonify(record.to_dict())
|
|
|
|
|
|
@sources_bp.route("", methods=["POST"])
|
|
async def create_source():
|
|
body = await request.get_json()
|
|
if not isinstance(body, dict):
|
|
return _bad("invalid_body", detail="body must be a JSON object")
|
|
try:
|
|
artist_id = int(body["artist_id"])
|
|
platform = body["platform"]
|
|
url = body["url"]
|
|
except (KeyError, TypeError, ValueError):
|
|
return _bad("invalid_body", detail="artist_id, platform, url are required")
|
|
|
|
optional = {
|
|
k: body[k]
|
|
for k in ("enabled", "config_overrides", "check_interval_override")
|
|
if k in body
|
|
}
|
|
|
|
async with get_session() as session:
|
|
svc = SourceService(session)
|
|
try:
|
|
record = await svc.create(
|
|
artist_id=artist_id, platform=platform, url=url, **optional,
|
|
)
|
|
except ArtistNotFoundError:
|
|
return _bad("artist_not_found", status=404, detail=f"artist id={artist_id}")
|
|
except UnknownPlatformError as exc:
|
|
return _bad("unknown_platform", detail=str(exc), known=sorted(KNOWN_PLATFORMS))
|
|
except InvalidConfigError as exc:
|
|
return _bad("invalid_config", detail=str(exc))
|
|
except EmptyUrlError as exc:
|
|
return _bad("empty_url", detail=str(exc))
|
|
except DuplicateSourceError as exc:
|
|
return _bad("duplicate", status=409, existing_id=exc.existing_id)
|
|
return jsonify(record.to_dict()), 201
|
|
|
|
|
|
@sources_bp.route("/<int:source_id>", methods=["PATCH"])
|
|
async def patch_source(source_id: int):
|
|
body = await request.get_json()
|
|
if not isinstance(body, dict):
|
|
return _bad("invalid_body", detail="body must be a JSON object")
|
|
async with get_session() as session:
|
|
svc = SourceService(session)
|
|
try:
|
|
record = await svc.update(source_id, **body)
|
|
except LookupError:
|
|
return _bad("not_found", status=404)
|
|
except UnknownPlatformError as exc:
|
|
return _bad("unknown_platform", detail=str(exc), known=sorted(KNOWN_PLATFORMS))
|
|
except InvalidConfigError as exc:
|
|
return _bad("invalid_config", detail=str(exc))
|
|
except EmptyUrlError as exc:
|
|
return _bad("empty_url", detail=str(exc))
|
|
except DuplicateSourceError as exc:
|
|
return _bad("duplicate", status=409, existing_id=exc.existing_id)
|
|
return jsonify(record.to_dict())
|
|
|
|
|
|
@sources_bp.route("/<int:source_id>", methods=["DELETE"])
|
|
async def delete_source(source_id: int):
|
|
async with get_session() as session:
|
|
try:
|
|
await SourceService(session).delete(source_id)
|
|
except LookupError:
|
|
return _bad("not_found", status=404)
|
|
return "", 204
|