Spaces:
Paused
Paused
File size: 4,910 Bytes
aceb1b2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """
Memos REST API (universal).
Blueprint mounted at /api/memos. Visibility/permission enforcement lives
in the service layer; this layer handles auth (logged-in user), the
feature gate (annotation_ui.memos), request parsing, and error mapping.
Privilege tier ("always read / may delete others"): adjudicators, via the
adjudication manager. Admin-dashboard memo moderation is a later follow-up.
"""
from __future__ import annotations
import logging
from functools import wraps
from flask import Blueprint, jsonify, request, session
from . import (
MemoError,
MemoNotFound,
MemoPermissionError,
create_memo,
delete_memo,
list_visible,
update_memo,
)
logger = logging.getLogger(__name__)
memos_bp = Blueprint("memos", __name__, url_prefix="/api/memos")
def _config() -> dict:
from potato.server_utils.config_module import config
return config
def memos_enabled(config: dict) -> bool:
"""Default: off in standard mode; on when qda_mode or solo_mode is on.
Explicit annotation_ui.memos always wins."""
ui = config.get("annotation_ui") or {}
if isinstance(ui, dict) and "memos" in ui:
return bool(ui["memos"])
return bool(
(config.get("qda_mode") or {}).get("enabled")
or (config.get("solo_mode") or {}).get("enabled")
)
def _default_visibility(config: dict) -> str:
ui = config.get("annotation_ui") or {}
v = ui.get("visibility") if isinstance(ui, dict) else None
return v if v in ("private", "shared") else "private"
def _is_privileged(username: str) -> bool:
try:
from potato.adjudication import get_adjudication_manager
adj = get_adjudication_manager()
return bool(adj and adj.is_adjudicator(username))
except Exception:
return False
def _ctx():
"""(task_dir, project, username, is_privileged) or None if not usable."""
config = _config()
if not memos_enabled(config):
return None, None, None, None, ("memos_disabled",)
username = session.get("username")
if not username:
return None, None, None, None, ("unauthenticated",)
task_dir = config.get("task_dir", ".")
project = config.get("annotation_task_name") or "default"
return task_dir, project, username, _is_privileged(username), None
def memos_required(view):
@wraps(view)
def wrapper(*args, **kwargs):
task_dir, project, username, priv, err = _ctx()
if err == ("memos_disabled",):
return jsonify({
"error": "Memos are not enabled in this deployment.",
"hint": "Set annotation_ui.memos: true (on by default in "
"qda_mode/solo_mode).",
}), 503
if err == ("unauthenticated",):
return jsonify({"error": "Not authenticated"}), 401
return view(task_dir, project, username, priv, *args, **kwargs)
return wrapper
def _handle(fn):
"""Map service exceptions to HTTP codes."""
try:
return fn()
except MemoNotFound as e:
return jsonify({"error": str(e)}), 404
except MemoPermissionError as e:
return jsonify({"error": str(e)}), 403
except MemoError as e:
return jsonify({"error": str(e)}), 400
@memos_bp.route("", methods=["GET"])
@memos_required
def list_memos(task_dir, project, username, priv):
instance_id = request.args.get("instance_id")
if not instance_id:
return jsonify({"error": "instance_id is required"}), 400
memos = list_visible(
task_dir, project=project, instance_id=instance_id,
requester=username, is_privileged=priv,
)
return jsonify({"memos": memos})
@memos_bp.route("", methods=["POST"])
@memos_required
def post_memo(task_dir, project, username, priv):
data = request.get_json(silent=True) or {}
instance_id = data.get("instance_id")
if not instance_id:
return jsonify({"error": "instance_id is required"}), 400
visibility = data.get("visibility") or _default_visibility(_config())
return _handle(lambda: jsonify({"memo": create_memo(
task_dir, project=project, instance_id=instance_id,
body=data.get("body", ""), created_by=username,
anchor=data.get("anchor"), visibility=visibility,
)}))
@memos_bp.route("/<memo_id>", methods=["PATCH"])
@memos_required
def patch_memo(task_dir, project, username, priv, memo_id):
data = request.get_json(silent=True) or {}
return _handle(lambda: jsonify({"memo": update_memo(
task_dir, memo_id, requester=username, is_privileged=priv,
body=data.get("body"), visibility=data.get("visibility"),
)}))
@memos_bp.route("/<memo_id>", methods=["DELETE"])
@memos_required
def remove_memo(task_dir, project, username, priv, memo_id):
def _do():
delete_memo(task_dir, memo_id, requester=username, is_privileged=priv)
return jsonify({"ok": True})
return _handle(_do)
|