bvandeusenandClaude Fable 5
64c641ce80
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / integration (push) Successful in 25s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Failing after 37s
CI & Build / Build & push image (push) Skipped
Reading the 28 route modules against each other and against the MCP tools: - routes/notes.py carried a PUT and a PATCH handler that were the same function minus the supersedes contract on one of them — one handler now serves both verbs, so both carry it. - The two _attach_supersession copies (REST + MCP) become supersession_svc.attach_relations(uid, note_id, data, hint=) — the seam the two surfaces must agree through; only the agent surface adds the one-sentence reading hint. - Three local _uid() wrappers over g.user.id → scribe.auth.get_current_user_id like every other module; design_systems' private _not_found → routes.utils. not_found; the four "********" literals → settings_svc.SECRET_MASK with the read/write contract written once. - routes/plugin.py: the project_id/repo resolution block and the comma-separated id parse were copied into three endpoints — _project_scope() and _int_list() now. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
313 lines
11 KiB
Python
313 lines
11 KiB
Python
"""Rulebook / topic REST endpoints.
|
|
|
|
Wraps services/rulebooks.py. Standard Scribe auth: get_current_user_id() is the
|
|
authenticated owner; the service enforces ownership scoping.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
|
|
from scribe.auth import get_current_user_id, login_required
|
|
import scribe.services.rulebooks as rulebooks_svc
|
|
from scribe.services.trash import delete as trash_delete
|
|
|
|
rulebooks_bp = Blueprint("rulebooks", __name__, url_prefix="/api")
|
|
|
|
|
|
|
|
# ── Rulebooks ───────────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.get("/rulebooks")
|
|
@login_required
|
|
async def list_rulebooks():
|
|
rows = await rulebooks_svc.list_rulebooks(get_current_user_id())
|
|
return jsonify({"rulebooks": [rb.to_dict() for rb in rows]})
|
|
|
|
|
|
@rulebooks_bp.post("/rulebooks")
|
|
@login_required
|
|
async def create_rulebook():
|
|
data = await request.get_json() or {}
|
|
title = (data.get("title") or "").strip()
|
|
if not title:
|
|
return jsonify({"error": "title is required"}), 400
|
|
rb = await rulebooks_svc.create_rulebook(
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
description=data.get("description", ""),
|
|
)
|
|
return jsonify(rb.to_dict()), 201
|
|
|
|
|
|
@rulebooks_bp.get("/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def get_rulebook(rulebook_id: int):
|
|
rb = await rulebooks_svc.get_rulebook(rulebook_id, get_current_user_id())
|
|
if rb is None:
|
|
return jsonify({"error": "rulebook not found"}), 404
|
|
return jsonify(rb.to_dict())
|
|
|
|
|
|
@rulebooks_bp.patch("/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def update_rulebook(rulebook_id: int):
|
|
data = await request.get_json() or {}
|
|
fields = {k: v for k, v in data.items() if k in ("title", "description", "always_on")}
|
|
rb = await rulebooks_svc.update_rulebook(rulebook_id, get_current_user_id(), **fields)
|
|
if rb is None:
|
|
return jsonify({"error": "rulebook not found"}), 404
|
|
return jsonify(rb.to_dict())
|
|
|
|
|
|
@rulebooks_bp.delete("/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def delete_rulebook(rulebook_id: int):
|
|
await trash_delete(get_current_user_id(), "rulebook", rulebook_id)
|
|
return "", 204
|
|
|
|
|
|
# ── Topics ──────────────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.get("/rulebooks/<int:rulebook_id>/topics")
|
|
@login_required
|
|
async def list_topics(rulebook_id: int):
|
|
try:
|
|
rows = await rulebooks_svc.list_topics(rulebook_id, get_current_user_id())
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify({"topics": [t.to_dict() for t in rows]})
|
|
|
|
|
|
@rulebooks_bp.post("/rulebooks/<int:rulebook_id>/topics")
|
|
@login_required
|
|
async def create_topic(rulebook_id: int):
|
|
data = await request.get_json() or {}
|
|
title = (data.get("title") or "").strip()
|
|
if not title:
|
|
return jsonify({"error": "title is required"}), 400
|
|
try:
|
|
topic = await rulebooks_svc.create_topic(
|
|
rulebook_id=rulebook_id,
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
description=data.get("description", ""),
|
|
order_index=data.get("order_index", 0),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify(topic.to_dict()), 201
|
|
|
|
|
|
@rulebooks_bp.patch("/rulebook-topics/<int:topic_id>")
|
|
@login_required
|
|
async def update_topic(topic_id: int):
|
|
data = await request.get_json() or {}
|
|
fields = {
|
|
k: v for k, v in data.items()
|
|
if k in ("title", "description", "order_index")
|
|
}
|
|
topic = await rulebooks_svc.update_topic(topic_id, get_current_user_id(), **fields)
|
|
if topic is None:
|
|
return jsonify({"error": "topic not found"}), 404
|
|
return jsonify(topic.to_dict())
|
|
|
|
|
|
@rulebooks_bp.delete("/rulebook-topics/<int:topic_id>")
|
|
@login_required
|
|
async def delete_topic(topic_id: int):
|
|
if await trash_delete(get_current_user_id(), "topic", topic_id) is None:
|
|
return jsonify({"error": "topic not found"}), 404
|
|
return "", 204
|
|
|
|
|
|
# ── Rules ───────────────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.get("/rules")
|
|
@login_required
|
|
async def list_rules():
|
|
def _opt_int(name):
|
|
raw = request.args.get(name)
|
|
return int(raw) if raw else None
|
|
|
|
try:
|
|
rulebook_id = _opt_int("rulebook_id")
|
|
topic_id = _opt_int("topic_id")
|
|
project_id = _opt_int("project_id")
|
|
except ValueError:
|
|
return jsonify({"error": "rulebook_id, topic_id, project_id must be integers"}), 400
|
|
|
|
rows = await rulebooks_svc.list_rules(
|
|
user_id=get_current_user_id(),
|
|
rulebook_id=rulebook_id,
|
|
topic_id=topic_id,
|
|
project_id=project_id,
|
|
)
|
|
return jsonify({"rules": [r.to_dict() for r in rows]})
|
|
|
|
|
|
@rulebooks_bp.post("/rulebook-topics/<int:topic_id>/rules")
|
|
@login_required
|
|
async def create_rule(topic_id: int):
|
|
data = await request.get_json() or {}
|
|
title = (data.get("title") or "").strip()
|
|
statement = (data.get("statement") or "").strip()
|
|
if not title or not statement:
|
|
return jsonify({"error": "title and statement are required"}), 400
|
|
try:
|
|
rule = await rulebooks_svc.create_rule(
|
|
topic_id=topic_id,
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
statement=statement,
|
|
why=data.get("why", ""),
|
|
how_to_apply=data.get("how_to_apply", ""),
|
|
order_index=data.get("order_index", 0),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify(rule.to_dict()), 201
|
|
|
|
|
|
@rulebooks_bp.get("/rules/<int:rule_id>")
|
|
@login_required
|
|
async def get_rule(rule_id: int):
|
|
rule = await rulebooks_svc.get_rule(rule_id, get_current_user_id())
|
|
if rule is None:
|
|
return jsonify({"error": "rule not found"}), 404
|
|
return jsonify(rule.to_dict())
|
|
|
|
|
|
@rulebooks_bp.patch("/rules/<int:rule_id>")
|
|
@login_required
|
|
async def update_rule(rule_id: int):
|
|
data = await request.get_json() or {}
|
|
fields = {
|
|
k: v for k, v in data.items()
|
|
if k in ("title", "statement", "why", "how_to_apply", "order_index")
|
|
}
|
|
rule = await rulebooks_svc.update_rule(rule_id, get_current_user_id(), **fields)
|
|
if rule is None:
|
|
return jsonify({"error": "rule not found"}), 404
|
|
return jsonify(rule.to_dict())
|
|
|
|
|
|
@rulebooks_bp.delete("/rules/<int:rule_id>")
|
|
@login_required
|
|
async def delete_rule(rule_id: int):
|
|
if await trash_delete(get_current_user_id(), "rule", rule_id) is None:
|
|
return jsonify({"error": "rule not found"}), 404
|
|
return "", 204
|
|
|
|
|
|
# ── Subscriptions ──────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/rulebook-subscriptions")
|
|
@login_required
|
|
async def subscribe_project(project_id: int):
|
|
data = await request.get_json() or {}
|
|
rulebook_id = data.get("rulebook_id")
|
|
if not rulebook_id:
|
|
return jsonify({"error": "rulebook_id is required"}), 400
|
|
try:
|
|
await rulebooks_svc.subscribe_project(
|
|
project_id=project_id, rulebook_id=int(rulebook_id), user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete(
|
|
"/projects/<int:project_id>/rulebook-subscriptions/<int:rulebook_id>"
|
|
)
|
|
@login_required
|
|
async def unsubscribe_project(project_id: int, rulebook_id: int):
|
|
try:
|
|
await rulebooks_svc.unsubscribe_project(
|
|
project_id=project_id, rulebook_id=rulebook_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.get("/projects/<int:project_id>/rules")
|
|
@login_required
|
|
async def get_project_rules(project_id: int):
|
|
result = await rulebooks_svc.get_applicable_rules(
|
|
project_id=project_id, user_id=get_current_user_id(),
|
|
)
|
|
return jsonify(result)
|
|
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/suppressions/rules/<int:rule_id>")
|
|
@login_required
|
|
async def suppress_project_rule(project_id: int, rule_id: int):
|
|
try:
|
|
await rulebooks_svc.suppress_rule_for_project(
|
|
project_id=project_id, rule_id=rule_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete("/projects/<int:project_id>/suppressions/rules/<int:rule_id>")
|
|
@login_required
|
|
async def unsuppress_project_rule(project_id: int, rule_id: int):
|
|
try:
|
|
await rulebooks_svc.unsuppress_rule_for_project(
|
|
project_id=project_id, rule_id=rule_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/suppressions/topics/<int:topic_id>")
|
|
@login_required
|
|
async def suppress_project_topic(project_id: int, topic_id: int):
|
|
try:
|
|
await rulebooks_svc.suppress_topic_for_project(
|
|
project_id=project_id, topic_id=topic_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete("/projects/<int:project_id>/suppressions/topics/<int:topic_id>")
|
|
@login_required
|
|
async def unsuppress_project_topic(project_id: int, topic_id: int):
|
|
try:
|
|
await rulebooks_svc.unsuppress_topic_for_project(
|
|
project_id=project_id, topic_id=topic_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/rules")
|
|
@login_required
|
|
async def create_project_rule(project_id: int):
|
|
"""Create a rule scoped to a single project. Frontend fast path."""
|
|
data = await request.get_json() or {}
|
|
statement = (data.get("statement") or "").strip()
|
|
if not statement:
|
|
return jsonify({"error": "statement is required"}), 400
|
|
title = (data.get("title") or "").strip() or statement.split(".")[0][:50]
|
|
try:
|
|
rule = await rulebooks_svc.create_project_rule(
|
|
project_id=project_id,
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
statement=statement,
|
|
why=data.get("why", ""),
|
|
how_to_apply=data.get("how_to_apply", ""),
|
|
order_index=data.get("order_index", 0),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify(rule.to_dict()), 201
|