Add tool rating & review from My Tools page

- Server: add GET /api/v1/tools/<owner>/<name>/my-review endpoint
- Client: add get_tool_rating, get_my_review, submit/update/delete review methods
- GUI: new ReviewDialog with star selector, title, content, submit/update/delete
- Tools page: rating display in details panel, Rate/Edit Rating button with
  context-sensitive enable/disable, background workers, rating cache
- Fix qualified name usage throughout tools page for owner-prefixed tools

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
rob 2026-01-30 02:15:23 -04:00
parent b00115a52e
commit 14b3f3d856
4 changed files with 756 additions and 8 deletions

View File

@ -0,0 +1,206 @@
"""Review dialog for rating registry tools."""
from typing import Optional, Dict
from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QLineEdit, QPlainTextEdit, QMessageBox
)
from PySide6.QtCore import Qt
class ReviewDialog(QDialog):
"""Dialog for submitting or editing a tool review."""
# Set when user clicks Delete - caller checks this after reject()
delete_requested = False
def __init__(
self,
parent,
owner: str,
name: str,
existing_review: Optional[Dict] = None
):
super().__init__(parent)
self._owner = owner
self._name = name
self._existing = existing_review
self._rating = existing_review["rating"] if existing_review else 0
self.delete_requested = False
self._setup_ui()
def _setup_ui(self):
editing = self._existing is not None
self.setWindowTitle(f"{'Edit Review' if editing else 'Rate Tool'} - {self._owner}/{self._name}")
self.setMinimumSize(420, 380)
layout = QVBoxLayout(self)
layout.setSpacing(14)
# Title
title = QLabel(f"{'Edit your review' if editing else 'Rate'} {self._owner}/{self._name}")
title.setStyleSheet("font-size: 15px; font-weight: 600;")
layout.addWidget(title)
# Star selector
star_label = QLabel("Rating:")
star_label.setStyleSheet("font-weight: 500;")
layout.addWidget(star_label)
star_row = QHBoxLayout()
star_row.setSpacing(4)
self._star_buttons = []
for i in range(1, 6):
btn = QPushButton("")
btn.setFixedSize(36, 36)
btn.setStyleSheet("""
QPushButton {
font-size: 20px;
border: 1px solid #e2e8f0;
border-radius: 4px;
background: #f7fafc;
}
QPushButton:hover {
background: #edf2f7;
}
""")
btn.setCursor(Qt.PointingHandCursor)
btn.clicked.connect(lambda checked, n=i: self._set_rating(n))
self._star_buttons.append(btn)
star_row.addWidget(btn)
star_row.addStretch()
layout.addLayout(star_row)
# Pre-fill rating
if self._rating:
self._set_rating(self._rating)
# Title input
title_label = QLabel("Title (optional):")
title_label.setStyleSheet("font-weight: 500;")
layout.addWidget(title_label)
self.title_input = QLineEdit()
self.title_input.setMaxLength(100)
self.title_input.setPlaceholderText("Brief summary of your review")
self.title_input.setMinimumHeight(28)
if self._existing and self._existing.get("title"):
self.title_input.setText(self._existing["title"])
layout.addWidget(self.title_input)
# Content input
content_label = QLabel("Review (optional, min 10 chars if provided):")
content_label.setStyleSheet("font-weight: 500;")
layout.addWidget(content_label)
self.content_input = QPlainTextEdit()
self.content_input.setPlaceholderText("Share your experience with this tool...")
self.content_input.setMinimumHeight(100)
self.content_input.setMaximumHeight(180)
if self._existing and self._existing.get("content"):
self.content_input.setPlainText(self._existing["content"])
layout.addWidget(self.content_input)
# Character count
self._char_count = QLabel("0 / 2000")
self._char_count.setStyleSheet("color: #718096; font-size: 11px;")
self._char_count.setAlignment(Qt.AlignRight)
self.content_input.textChanged.connect(self._update_char_count)
self._update_char_count()
layout.addWidget(self._char_count)
layout.addStretch()
# Buttons
btn_layout = QHBoxLayout()
if editing:
self.btn_delete = QPushButton("Delete Review")
self.btn_delete.setObjectName("danger")
self.btn_delete.clicked.connect(self._on_delete)
btn_layout.addWidget(self.btn_delete)
btn_layout.addStretch()
btn_cancel = QPushButton("Cancel")
btn_cancel.setObjectName("secondary")
btn_cancel.clicked.connect(self.reject)
btn_layout.addWidget(btn_cancel)
submit_text = "Update" if editing else "Submit"
self.btn_submit = QPushButton(submit_text)
self.btn_submit.clicked.connect(self._on_submit)
btn_layout.addWidget(self.btn_submit)
layout.addLayout(btn_layout)
def _set_rating(self, n: int):
self._rating = n
for i, btn in enumerate(self._star_buttons):
if i < n:
btn.setText("")
btn.setStyleSheet("""
QPushButton {
font-size: 20px;
border: 1px solid #d69e2e;
border-radius: 4px;
background: #fefcbf;
color: #d69e2e;
}
QPushButton:hover { background: #fef3c7; }
""")
else:
btn.setText("")
btn.setStyleSheet("""
QPushButton {
font-size: 20px;
border: 1px solid #e2e8f0;
border-radius: 4px;
background: #f7fafc;
}
QPushButton:hover { background: #edf2f7; }
""")
def _update_char_count(self):
count = len(self.content_input.toPlainText())
self._char_count.setText(f"{count} / 2000")
if count > 2000:
self._char_count.setStyleSheet("color: #e53e3e; font-size: 11px;")
else:
self._char_count.setStyleSheet("color: #718096; font-size: 11px;")
def _on_submit(self):
if self._rating < 1:
QMessageBox.warning(self, "Validation", "Please select a rating (1-5 stars).")
return
content = self.content_input.toPlainText().strip()
if content and len(content) < 10:
QMessageBox.warning(self, "Validation", "Review content must be at least 10 characters.")
return
if len(content) > 2000:
QMessageBox.warning(self, "Validation", "Review content must be 2000 characters or less.")
return
self.accept()
def _on_delete(self):
reply = QMessageBox.question(
self, "Delete Review",
"Are you sure you want to delete your review?",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No
)
if reply == QMessageBox.Yes:
self.delete_requested = True
self.reject()
def get_review_data(self) -> Dict:
"""Get the review data entered by the user."""
return {
"rating": self._rating,
"title": self.title_input.text().strip(),
"content": self.content_input.toPlainText().strip()[:2000],
}

View File

@ -2,7 +2,7 @@
from collections import defaultdict from collections import defaultdict
from pathlib import Path from pathlib import Path
from typing import Optional, Tuple from typing import Optional, Tuple, Dict, Any
import yaml import yaml
@ -154,6 +154,138 @@ def get_tool_publish_state(tool_name: str) -> Tuple[str, Optional[str]]:
return ("local", None) return ("local", None)
def get_tool_registry_info(tool_name: str) -> Optional[Tuple[str, str]]:
"""
Get registry owner/name for a tool if it's registry-sourced.
Returns:
(owner, name) tuple if tool is from registry, None if local-only.
"""
tools_dir = get_tools_dir()
config_path = tools_dir / tool_name / "config.yaml"
if not config_path.exists():
return None
try:
config_data = yaml.safe_load(config_path.read_text()) or {}
if not config_data.get("registry_hash"):
return None
# Check if stored under owner subdir: ~/.cmdforge/<owner>/<name>/
if "/" in tool_name:
owner, name = tool_name.split("/", 1)
return (owner, name)
# Flat dir: check source.author or installed_from in config
source = config_data.get("source", {})
if isinstance(source, dict) and source.get("author"):
return (source["author"], tool_name)
if config_data.get("installed_from"):
parts = config_data["installed_from"].split("/", 1)
if len(parts) == 2:
return (parts[0], parts[1])
return None
except Exception:
return None
class RatingWorker(QThread):
"""Background worker to fetch rating summary and user's review for a tool."""
finished = Signal(str, dict) # tool_name, result dict
error = Signal(str, str) # tool_name, error message
def __init__(self, tool_name: str, owner: str, name: str):
super().__init__()
self.tool_name = tool_name
self.owner = owner
self.name = name
def run(self):
from ...registry_client import RegistryClient, RegistryError
try:
config = load_config()
client = RegistryClient()
client.token = config.registry.token
result: Dict[str, Any] = {"rating": None, "my_review": None}
try:
result["rating"] = client.get_tool_rating(self.owner, self.name)
except RegistryError:
pass
if client.token:
try:
result["my_review"] = client.get_my_review(self.owner, self.name)
except RegistryError:
pass
self.finished.emit(self.tool_name, result)
except Exception as e:
self.error.emit(self.tool_name, str(e))
class SubmitReviewWorker(QThread):
"""Background worker to submit or update a review."""
finished = Signal(str) # success message
error = Signal(str) # error message
def __init__(self, owner: str, name: str, rating: int, title: str, content: str,
review_id: Optional[int] = None):
super().__init__()
self.owner = owner
self.name = name
self.rating = rating
self.title = title
self.content = content
self.review_id = review_id # None = new, int = update
def run(self):
from ...registry_client import RegistryClient, RegistryError, RateLimitError
try:
config = load_config()
client = RegistryClient()
client.token = config.registry.token
if self.review_id is not None:
client.update_review(self.review_id, self.rating, self.title, self.content)
self.finished.emit("Review updated successfully")
else:
client.submit_review(self.owner, self.name, self.rating, self.title, self.content)
self.finished.emit("Review submitted successfully")
except RateLimitError as e:
minutes = max(1, e.retry_after // 60)
self.error.emit(f"Too many reviews. Try again in {minutes} minute(s).")
except RegistryError as e:
self.error.emit(e.message)
except Exception as e:
self.error.emit(str(e))
class DeleteReviewWorker(QThread):
"""Background worker to delete a review."""
finished = Signal(str)
error = Signal(str)
def __init__(self, review_id: int):
super().__init__()
self.review_id = review_id
def run(self):
from ...registry_client import RegistryClient, RegistryError
try:
config = load_config()
client = RegistryClient()
client.token = config.registry.token
client.delete_review(self.review_id)
self.finished.emit("Review deleted successfully")
except RegistryError as e:
self.error.emit(e.message)
except Exception as e:
self.error.emit(str(e))
class ToolsPage(QWidget): class ToolsPage(QWidget):
"""Main tools management page.""" """Main tools management page."""
@ -165,6 +297,11 @@ class ToolsPage(QWidget):
self._syncing = False # Prevent re-sync during update self._syncing = False # Prevent re-sync during update
self._poll_timer = None # Timer for automatic status polling self._poll_timer = None # Timer for automatic status polling
self._has_pending_tools = False # Track if we need to poll self._has_pending_tools = False # Track if we need to poll
self._rating_cache: Dict[str, Dict] = {} # tool_name -> {rating, my_review}
self._rating_worker = None
self._review_worker = None
self._my_slug: Optional[str] = None # Cached current user slug
self._my_slug_fetched = False
self._setup_ui() self._setup_ui()
self.refresh() self.refresh()
@ -264,6 +401,13 @@ class ToolsPage(QWidget):
self.btn_delete.setEnabled(False) self.btn_delete.setEnabled(False)
btn_layout.addWidget(self.btn_delete) btn_layout.addWidget(self.btn_delete)
self.btn_rate = QPushButton("Rate Tool")
self.btn_rate.setObjectName("secondary")
self.btn_rate.clicked.connect(self._rate_tool)
self.btn_rate.setEnabled(False)
self.btn_rate.setToolTip("Local tools can't be rated")
btn_layout.addWidget(self.btn_rate)
btn_layout.addStretch() btn_layout.addStretch()
# Connect/Publish button # Connect/Publish button
@ -498,7 +642,8 @@ class ToolsPage(QWidget):
tool = load_tool(tool_name) tool = load_tool(tool_name)
if tool: if tool:
self._current_tool = tool self._current_tool = tool
self._show_tool_info(tool) self._show_tool_info(tool, tool_name)
self._fetch_rating_if_needed(tool_name)
self._update_buttons() self._update_buttons()
def _get_qualified_name(self) -> Optional[str]: def _get_qualified_name(self) -> Optional[str]:
@ -525,8 +670,15 @@ class ToolsPage(QWidget):
pass pass
return None return None
def _show_tool_info(self, tool: Tool): def _show_tool_info(self, tool: Tool, qualified_name: Optional[str] = None):
"""Display tool information.""" """Display tool information.
Args:
tool: The Tool object.
qualified_name: The qualified name (e.g. 'official/foo') used for
config path lookups. Falls back to tool.name if not provided.
"""
qname = qualified_name or tool.name
lines = [] lines = []
# Name and description # Name and description
@ -535,7 +687,7 @@ class ToolsPage(QWidget):
lines.append(f"<p style='color: #4a5568; margin-bottom: 16px;'>{tool.description}</p>") lines.append(f"<p style='color: #4a5568; margin-bottom: 16px;'>{tool.description}</p>")
# Publish state # Publish state
state, registry_hash = get_tool_publish_state(tool.name) state, registry_hash = get_tool_publish_state(qname)
if state == "published": if state == "published":
lines.append( lines.append(
"<p style='background: #c6f6d5; color: #276749; padding: 6px 10px; " "<p style='background: #c6f6d5; color: #276749; padding: 6px 10px; "
@ -561,7 +713,7 @@ class ToolsPage(QWidget):
"⚠ Changes requested - please address feedback and republish</p>" "⚠ Changes requested - please address feedback and republish</p>"
) )
# Show feedback if available # Show feedback if available
feedback = self._get_tool_feedback(tool.name) feedback = self._get_tool_feedback(qname)
if feedback: if feedback:
feedback_escaped = feedback.replace("<", "&lt;").replace(">", "&gt;").replace("\n", "<br>") feedback_escaped = feedback.replace("<", "&lt;").replace(">", "&gt;").replace("\n", "<br>")
lines.append( lines.append(
@ -576,7 +728,7 @@ class ToolsPage(QWidget):
"✗ Rejected by moderator</p>" "✗ Rejected by moderator</p>"
) )
# Show feedback if available # Show feedback if available
feedback = self._get_tool_feedback(tool.name) feedback = self._get_tool_feedback(qname)
if feedback: if feedback:
feedback_escaped = feedback.replace("<", "&lt;").replace(">", "&gt;").replace("\n", "<br>") feedback_escaped = feedback.replace("<", "&lt;").replace(">", "&gt;").replace("\n", "<br>")
lines.append( lines.append(
@ -585,6 +737,43 @@ class ToolsPage(QWidget):
f"<strong>Reason:</strong><br>{feedback_escaped}</div>" f"<strong>Reason:</strong><br>{feedback_escaped}</div>"
) )
# Rating section (for registry-sourced tools)
registry_info = get_tool_registry_info(qname)
if registry_info:
cached = self._rating_cache.get(qname)
if cached:
rating_data = cached.get("rating")
my_review = cached.get("my_review")
if rating_data:
avg = rating_data.get("average_rating", 0)
count = rating_data.get("rating_count", 0)
if count > 0:
filled = round(avg)
stars_html = "".join(
"" if i < filled else "" for i in range(5)
)
lines.append(
f"<p style='color: #d69e2e; font-size: 14px; margin-bottom: 4px;'>"
f"{stars_html} <span style='color: #4a5568; font-size: 12px;'>"
f"{avg:.1f}/5 from {count} review{'s' if count != 1 else ''}</span></p>"
)
else:
lines.append(
"<p style='color: #718096; font-size: 12px; margin-bottom: 4px;'>"
"No ratings yet</p>"
)
if my_review:
my_stars = "" * my_review["rating"] + "" * (5 - my_review["rating"])
lines.append(
f"<p style='color: #805ad5; font-size: 12px; margin-bottom: 8px;'>"
f"Your rating: {my_stars}</p>"
)
elif registry_info:
lines.append(
"<p style='color: #718096; font-size: 12px; margin-bottom: 4px;'>"
"Loading ratings...</p>"
)
# Source info # Source info
if tool.source: if tool.source:
source_type = tool.source.type source_type = tool.source.type
@ -654,6 +843,9 @@ class ToolsPage(QWidget):
# Not connected - Connect button is always enabled # Not connected - Connect button is always enabled
self.btn_publish.setEnabled(True) self.btn_publish.setEnabled(True)
# Rate button state
self._update_rate_button()
def _create_tool(self): def _create_tool(self):
"""Create a new tool.""" """Create a new tool."""
self.main_window.open_tool_builder() self.main_window.open_tool_builder()
@ -734,12 +926,196 @@ class ToolsPage(QWidget):
if result: if result:
self.main_window.show_status(f"Published '{tool_name}'") self.main_window.show_status(f"Published '{tool_name}'")
def _update_rate_button(self):
"""Update the Rate button state based on current selection."""
if not self._current_tool:
self.btn_rate.setEnabled(False)
self.btn_rate.setText("Rate Tool")
self.btn_rate.setToolTip("Select a tool to rate")
return
tool_name = self._get_qualified_name()
if not tool_name:
self.btn_rate.setEnabled(False)
self.btn_rate.setText("Rate Tool")
self.btn_rate.setToolTip("Select a tool to rate")
return
registry_info = get_tool_registry_info(tool_name)
if not registry_info:
self.btn_rate.setEnabled(False)
self.btn_rate.setText("Rate Tool")
self.btn_rate.setToolTip("Local tools can't be rated")
return
config = load_config()
if not config.registry.token:
self.btn_rate.setEnabled(False)
self.btn_rate.setText("Rate Tool")
self.btn_rate.setToolTip("Sign in to rate")
return
owner, name = registry_info
# Check if this is the user's own tool
if self._my_slug and owner == self._my_slug:
self.btn_rate.setEnabled(False)
self.btn_rate.setText("Rate Tool")
self.btn_rate.setToolTip("You can't rate your own tool")
return
# Check if user has an existing review
cached = self._rating_cache.get(tool_name)
if cached and cached.get("my_review"):
self.btn_rate.setText("Edit Rating")
else:
self.btn_rate.setText("Rate Tool")
self.btn_rate.setEnabled(True)
self.btn_rate.setToolTip("")
def _fetch_my_slug(self):
"""Fetch and cache the current user's slug."""
if self._my_slug_fetched:
return
self._my_slug_fetched = True
config = load_config()
if not config.registry.token:
return
try:
from ...registry_client import RegistryClient, RegistryError
client = RegistryClient()
client.token = config.registry.token
me = client.get_me()
self._my_slug = me.get("slug")
except Exception:
pass
def _fetch_rating_if_needed(self, tool_name: str):
"""Fetch rating data for a registry tool if not cached."""
registry_info = get_tool_registry_info(tool_name)
if not registry_info:
return
# Ensure we have the user slug (non-blocking first time; runs in main thread once)
if not self._my_slug_fetched:
self._fetch_my_slug()
if tool_name in self._rating_cache:
# Already cached - just update buttons
self._update_rate_button()
return
owner, name = registry_info
# Stop any existing rating worker
if self._rating_worker and self._rating_worker.isRunning():
self._rating_worker.wait(500)
self._rating_worker = RatingWorker(tool_name, owner, name)
self._rating_worker.finished.connect(self._on_rating_fetched)
self._rating_worker.error.connect(self._on_rating_error)
self._rating_worker.start()
def _on_rating_fetched(self, tool_name: str, result: Dict):
"""Handle rating data fetched from registry."""
self._rating_cache[tool_name] = result
# If this is still the selected tool, refresh its display
qualified = self._get_qualified_name()
current_name = qualified or (self._current_tool.name if self._current_tool else None)
if current_name == tool_name and self._current_tool:
self._show_tool_info(self._current_tool, tool_name)
self._update_rate_button()
def _on_rating_error(self, tool_name: str, error: str):
"""Handle rating fetch error (non-critical)."""
pass
def _rate_tool(self):
"""Open the review dialog for the selected tool."""
if not self._current_tool:
return
tool_name = self._get_qualified_name() or self._current_tool.name
registry_info = get_tool_registry_info(tool_name)
if not registry_info:
return
owner, name = registry_info
cached = self._rating_cache.get(tool_name, {})
existing_review = cached.get("my_review")
from ..dialogs.review_dialog import ReviewDialog
dialog = ReviewDialog(self, owner, name, existing_review)
result = dialog.exec()
if dialog.delete_requested and existing_review:
# User wants to delete their review
self._delete_review(tool_name, existing_review["id"])
elif result:
# User submitted/updated
data = dialog.get_review_data()
review_id = existing_review["id"] if existing_review else None
self._submit_review(tool_name, owner, name, data, review_id)
def _submit_review(self, tool_name: str, owner: str, name: str,
data: Dict, review_id: Optional[int]):
"""Submit or update a review in background."""
if self._review_worker and self._review_worker.isRunning():
self._review_worker.wait(1000)
self._review_worker = SubmitReviewWorker(
owner, name, data["rating"], data["title"], data["content"],
review_id=review_id
)
self._review_worker.finished.connect(
lambda msg: self._on_review_action_done(tool_name, msg)
)
self._review_worker.error.connect(
lambda err: self._on_review_action_error(err)
)
self._review_worker.start()
def _delete_review(self, tool_name: str, review_id: int):
"""Delete a review in background."""
if self._review_worker and self._review_worker.isRunning():
self._review_worker.wait(1000)
self._review_worker = DeleteReviewWorker(review_id)
self._review_worker.finished.connect(
lambda msg: self._on_review_action_done(tool_name, msg)
)
self._review_worker.error.connect(
lambda err: self._on_review_action_error(err)
)
self._review_worker.start()
def _on_review_action_done(self, tool_name: str, message: str):
"""Handle review submit/update/delete success."""
self.main_window.show_status(message)
# Invalidate cache and re-fetch
self._rating_cache.pop(tool_name, None)
qualified = self._get_qualified_name()
current_name = qualified or (self._current_tool.name if self._current_tool else None)
if current_name == tool_name:
self._fetch_rating_if_needed(tool_name)
def _on_review_action_error(self, error: str):
"""Handle review action error."""
self.main_window.show_status(f"Review error: {error}")
def _sync_tool_status(self): def _sync_tool_status(self):
"""Sync the moderation status of the selected tool from the registry.""" """Sync the moderation status of the selected tool from the registry."""
if not self._current_tool: if not self._current_tool:
return return
tool_name = self._current_tool.name tool_name = self._get_qualified_name() or self._current_tool.name
config_path = get_tools_dir() / tool_name / "config.yaml" config_path = get_tools_dir() / tool_name / "config.yaml"
if not config_path.exists(): if not config_path.exists():

View File

@ -4318,6 +4318,40 @@ def create_app() -> Flask:
# ─── Reviews & Ratings API ──────────────────────────────────────────────────── # ─── Reviews & Ratings API ────────────────────────────────────────────────────
@app.route("/api/v1/tools/<owner>/<name>/my-review", methods=["GET"])
@require_token
def get_my_review(owner: str, name: str) -> Response:
"""Get the current user's review for a tool."""
if not OWNER_RE.match(owner) or not TOOL_NAME_RE.match(name):
return error_response("VALIDATION_ERROR", "Invalid owner or tool name")
tool = query_one(
g.db,
"SELECT id FROM tools WHERE owner = ? AND name = ? ORDER BY id DESC LIMIT 1",
[owner, name],
)
if not tool:
return error_response("TOOL_NOT_FOUND", f"Tool '{owner}/{name}' not found", 404)
review = query_one(
g.db,
"SELECT id, rating, title, content, created_at, updated_at FROM reviews WHERE tool_id = ? AND reviewer_id = ?",
[tool["id"], g.current_publisher["id"]],
)
if not review:
return error_response("REVIEW_NOT_FOUND", "No review found", 404)
return jsonify({
"data": {
"id": review["id"],
"rating": review["rating"],
"title": review["title"],
"content": review["content"],
"created_at": review["created_at"],
"updated_at": review["updated_at"],
}
})
@app.route("/api/v1/tools/<owner>/<name>/reviews", methods=["POST"]) @app.route("/api/v1/tools/<owner>/<name>/reviews", methods=["POST"])
@require_token @require_token
def submit_review(owner: str, name: str) -> Response: def submit_review(owner: str, name: str) -> Response:

View File

@ -821,6 +821,138 @@ class RegistryClient:
return response.json().get("data", {}) return response.json().get("data", {})
# -------------------------------------------------------------------------
# Reviews & Ratings
# -------------------------------------------------------------------------
def get_tool_rating(self, owner: str, name: str) -> Dict[str, Any]:
"""
Get rating summary for a tool.
Args:
owner: Tool owner
name: Tool name
Returns:
Dict with average_rating, rating_count, distribution, etc.
"""
response = self._request("GET", f"/tools/{owner}/{name}/rating")
if response.status_code == 404:
raise RegistryError(
code="TOOL_NOT_FOUND",
message=f"Tool '{owner}/{name}' not found",
http_status=404
)
if response.status_code != 200:
self._handle_error_response(response)
return response.json().get("data", {})
def get_my_review(self, owner: str, name: str) -> Optional[Dict[str, Any]]:
"""
Get the current user's review for a tool.
Args:
owner: Tool owner
name: Tool name
Returns:
Review dict or None if no review exists
"""
response = self._request("GET", f"/tools/{owner}/{name}/my-review", require_auth=True)
if response.status_code == 404:
# No review found - not an error
return None
if response.status_code != 200:
self._handle_error_response(response)
return response.json().get("data", {})
def submit_review(
self, owner: str, name: str, rating: int, title: str = "", content: str = ""
) -> Dict[str, Any]:
"""
Submit a review for a tool.
Args:
owner: Tool owner
name: Tool name
rating: Rating 1-5
title: Optional review title
content: Optional review content
Returns:
Dict with review id and data
"""
payload: Dict[str, Any] = {"rating": rating}
if title:
payload["title"] = title
if content:
payload["content"] = content
response = self._request(
"POST", f"/tools/{owner}/{name}/reviews",
json_data=payload,
require_auth=True
)
if response.status_code not in (200, 201):
self._handle_error_response(response)
return response.json().get("data", {})
def update_review(
self, review_id: int, rating: int, title: str = "", content: str = ""
) -> Dict[str, Any]:
"""
Update an existing review.
Args:
review_id: Review ID
rating: Rating 1-5
title: Review title
content: Review content
Returns:
Dict with status
"""
payload: Dict[str, Any] = {"rating": rating, "title": title, "content": content}
response = self._request(
"PUT", f"/reviews/{review_id}",
json_data=payload,
require_auth=True
)
if response.status_code != 200:
self._handle_error_response(response)
return response.json().get("data", {})
def delete_review(self, review_id: int) -> Dict[str, Any]:
"""
Delete a review.
Args:
review_id: Review ID
Returns:
Dict with status
"""
response = self._request(
"DELETE", f"/reviews/{review_id}",
require_auth=True
)
if response.status_code != 200:
self._handle_error_response(response)
return response.json().get("data", {})
def get_popular_tools(self, limit: int = 10) -> List[ToolInfo]: def get_popular_tools(self, limit: int = 10) -> List[ToolInfo]:
""" """
Get most popular tools. Get most popular tools.