rob
/
CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects 1 Releases Wiki Activity

Fix NameError in admin collections endpoints 

- Replace undefined get_current_user() with existing g.current_publisher
- Replace undefined log_admin_action() with existing log_audit() function
- Remove redundant role checks (already handled by @require_admin decorator)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
rob 2026-01-17 04:06:19 -04:00
parent c35512dee9
commit 2b927478a6
1 changed files with 4 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/cmdforge/registry/app.py
View File

@ -1360,10 +1360,6 @@ def create_app() -> Flask:
@require_admin @require_admin
def admin_list_collections() -> Response: def admin_list_collections() -> Response:
"""List all collections with full details (admin).""" """List all collections with full details (admin)."""
user = get_current_user()
if not user or user.get("role") not in ("admin", "moderator"):
return error_response("FORBIDDEN", "Admin access required", 403)
rows = query_all( rows = query_all(
g.db, g.db,
"SELECT * FROM collections ORDER BY name", "SELECT * FROM collections ORDER BY name",
@ -1392,16 +1388,12 @@ def create_app() -> Flask:
@require_admin @require_admin
def admin_create_collection() -> Response: def admin_create_collection() -> Response:
"""Create a new collection (admin).""" """Create a new collection (admin)."""
user = get_current_user()
if not user or user.get("role") != "admin":
return error_response("FORBIDDEN", "Admin access required", 403)
data = request.get_json() or {} data = request.get_json() or {}
name = data.get("name", "").strip().lower() name = data.get("name", "").strip().lower()
display_name = data.get("display_name", "").strip() display_name = data.get("display_name", "").strip()
description = data.get("description", "").strip() description = data.get("description", "").strip()
icon = data.get("icon", "").strip() icon = data.get("icon", "").strip()
maintainer = data.get("maintainer", user.get("username", "")).strip() maintainer = data.get("maintainer", g.current_publisher.get("slug", "")).strip()
tools = data.get("tools", []) tools = data.get("tools", [])
pinned = data.get("pinned", {}) pinned = data.get("pinned", {})
tags = data.get("tags", []) tags = data.get("tags", [])
@ -1441,7 +1433,7 @@ def create_app() -> Flask:
g.db.commit() g.db.commit()
# Audit log # Audit log
log_admin_action(g.db, user["id"], "create_collection", {"collection": name}) log_audit("create_collection", "collection", name, {"collection": name})
return jsonify({ return jsonify({
"success": True, "success": True,
@ -1456,10 +1448,6 @@ def create_app() -> Flask:
@require_admin @require_admin
def admin_update_collection(name: str) -> Response: def admin_update_collection(name: str) -> Response:
"""Update a collection (admin).""" """Update a collection (admin)."""
user = get_current_user()
if not user or user.get("role") != "admin":
return error_response("FORBIDDEN", "Admin access required", 403)
existing = query_one(g.db, "SELECT * FROM collections WHERE name = ?", [name]) existing = query_one(g.db, "SELECT * FROM collections WHERE name = ?", [name])
if not existing: if not existing:
return error_response("COLLECTION_NOT_FOUND", f"Collection '{name}' not found", 404) return error_response("COLLECTION_NOT_FOUND", f"Collection '{name}' not found", 404)
@ -1512,7 +1500,7 @@ def create_app() -> Flask:
g.db.commit() g.db.commit()
# Audit log # Audit log
log_admin_action(g.db, user["id"], "update_collection", {"collection": name}) log_audit("update_collection", "collection", name, {"collection": name})
return jsonify({ return jsonify({
"success": True, "success": True,
@ -1526,10 +1514,6 @@ def create_app() -> Flask:
@require_admin @require_admin
def admin_delete_collection(name: str) -> Response: def admin_delete_collection(name: str) -> Response:
"""Delete a collection (admin).""" """Delete a collection (admin)."""
user = get_current_user()
if not user or user.get("role") != "admin":
return error_response("FORBIDDEN", "Admin access required", 403)
existing = query_one(g.db, "SELECT id FROM collections WHERE name = ?", [name]) existing = query_one(g.db, "SELECT id FROM collections WHERE name = ?", [name])
if not existing: if not existing:
return error_response("COLLECTION_NOT_FOUND", f"Collection '{name}' not found", 404) return error_response("COLLECTION_NOT_FOUND", f"Collection '{name}' not found", 404)
@ -1539,7 +1523,7 @@ def create_app() -> Flask:
g.db.commit() g.db.commit()
# Audit log # Audit log
log_admin_action(g.db, user["id"], "delete_collection", {"collection": name}) log_audit("delete_collection", "collection", name, {"collection": name})
return jsonify({ return jsonify({
"success": True, "success": True,