Admin routes and the SPA fallback answer errors through responses.py
responses.py is the app's one JSON error shape, yet invites, accounts, the SPA
fallback and the test-email route still built jsonify({"error": ...}) by hand,
and two parsed path ids with their own try/uuid.UUID. They now use json_error,
not_found and parse_uuid. The download limiter's 429 stays for F13, with its
Retry-After twin.
DRY pass #2, batch 4, F10 (#5372).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -6,7 +6,6 @@ What a reset link is, and how it is used, is in `password_resets.py`.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import uuid
|
|
||||||
|
|
||||||
from quart import Blueprint, g, jsonify, request
|
from quart import Blueprint, g, jsonify, request
|
||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
@@ -18,6 +17,7 @@ from .db import session_scope
|
|||||||
from .models.user import User
|
from .models.user import User
|
||||||
from .password_resets import issue
|
from .password_resets import issue
|
||||||
from .proxy import client_address
|
from .proxy import client_address
|
||||||
|
from .responses import not_found, parse_uuid
|
||||||
|
|
||||||
bp = Blueprint("accounts", __name__, url_prefix="/api/accounts")
|
bp = Blueprint("accounts", __name__, url_prefix="/api/accounts")
|
||||||
|
|
||||||
@@ -46,14 +46,13 @@ async def list_accounts():
|
|||||||
@bp.post("/<account_id>/reset-link")
|
@bp.post("/<account_id>/reset-link")
|
||||||
@require_admin
|
@require_admin
|
||||||
async def create_reset_link(account_id: str):
|
async def create_reset_link(account_id: str):
|
||||||
try:
|
uid = parse_uuid(account_id)
|
||||||
uid = uuid.UUID(account_id)
|
if uid is None:
|
||||||
except ValueError:
|
return not_found()
|
||||||
return jsonify({"error": "not found"}), 404
|
|
||||||
async with session_scope() as db:
|
async with session_scope() as db:
|
||||||
user = await db.get(User, uid)
|
user = await db.get(User, uid)
|
||||||
if user is None:
|
if user is None:
|
||||||
return jsonify({"error": "not found"}), 404
|
return not_found()
|
||||||
token, expires_at = await issue(db, uid, g.user_id)
|
token, expires_at = await issue(db, uid, g.user_id)
|
||||||
await db.commit()
|
await db.commit()
|
||||||
email = user.email
|
email = user.email
|
||||||
|
|||||||
+3
-2
@@ -23,6 +23,7 @@ from .invites_api import bp as invites_bp
|
|||||||
from .labels import bp as labels_bp
|
from .labels import bp as labels_bp
|
||||||
from .notes import bp as notes_bp
|
from .notes import bp as notes_bp
|
||||||
from .proxy import is_https
|
from .proxy import is_https
|
||||||
|
from .responses import json_error, not_found
|
||||||
from .retention import run_sweeper
|
from .retention import run_sweeper
|
||||||
from .settings import MAX_BODY_MB, get_public_config, get_setting, load_or_create_secret_key, refresh_live
|
from .settings import MAX_BODY_MB, get_public_config, get_setting, load_or_create_secret_key, refresh_live
|
||||||
from .settings_api import bp as settings_bp
|
from .settings_api import bp as settings_bp
|
||||||
@@ -225,13 +226,13 @@ def create_app() -> Quart:
|
|||||||
@app.get("/<path:path>")
|
@app.get("/<path:path>")
|
||||||
async def spa(path: str):
|
async def spa(path: str):
|
||||||
if path.startswith("api/"):
|
if path.startswith("api/"):
|
||||||
return jsonify({"error": "not found"}), 404
|
return not_found()
|
||||||
candidate = os.path.join(STATIC_DIR, path)
|
candidate = os.path.join(STATIC_DIR, path)
|
||||||
if path and os.path.isfile(candidate):
|
if path and os.path.isfile(candidate):
|
||||||
return await send_from_directory(STATIC_DIR, path)
|
return await send_from_directory(STATIC_DIR, path)
|
||||||
index = os.path.join(STATIC_DIR, "index.html")
|
index = os.path.join(STATIC_DIR, "index.html")
|
||||||
if os.path.isfile(index):
|
if os.path.isfile(index):
|
||||||
return await send_from_directory(STATIC_DIR, "index.html")
|
return await send_from_directory(STATIC_DIR, "index.html")
|
||||||
return jsonify({"error": "frontend not built"}), 404
|
return json_error("frontend not built", 404)
|
||||||
|
|
||||||
return app
|
return app
|
||||||
|
|||||||
@@ -5,7 +5,6 @@ What an invite is, and how it is redeemed, is in `invites.py`.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import uuid
|
|
||||||
from datetime import datetime, timedelta, timezone
|
from datetime import datetime, timedelta, timezone
|
||||||
|
|
||||||
from quart import Blueprint, g, jsonify, request
|
from quart import Blueprint, g, jsonify, request
|
||||||
@@ -19,6 +18,7 @@ from .invites import MAX_DAYS, lifetime_days, serialize
|
|||||||
from .models.invite import Invite
|
from .models.invite import Invite
|
||||||
from .models.user import User
|
from .models.user import User
|
||||||
from .proxy import client_address
|
from .proxy import client_address
|
||||||
|
from .responses import json_error, not_found, parse_uuid
|
||||||
from .security import generate_token, hash_token
|
from .security import generate_token, hash_token
|
||||||
|
|
||||||
bp = Blueprint("invites", __name__, url_prefix="/api/invites")
|
bp = Blueprint("invites", __name__, url_prefix="/api/invites")
|
||||||
@@ -33,10 +33,10 @@ async def create_invite():
|
|||||||
data = await request.get_json(silent=True) or {}
|
data = await request.get_json(silent=True) or {}
|
||||||
email = (data.get("email") or "").strip().lower() or None
|
email = (data.get("email") or "").strip().lower() or None
|
||||||
if email is not None and "@" not in email:
|
if email is not None and "@" not in email:
|
||||||
return jsonify({"error": "that doesn't look like an email address"}), 400
|
return json_error("that doesn't look like an email address", 400)
|
||||||
days = lifetime_days(data.get("days"))
|
days = lifetime_days(data.get("days"))
|
||||||
if days is None:
|
if days is None:
|
||||||
return jsonify({"error": f"an invite lasts between 1 and {MAX_DAYS} days"}), 400
|
return json_error(f"an invite lasts between 1 and {MAX_DAYS} days", 400)
|
||||||
|
|
||||||
token = generate_token()
|
token = generate_token()
|
||||||
now = datetime.now(timezone.utc)
|
now = datetime.now(timezone.utc)
|
||||||
@@ -82,15 +82,14 @@ async def list_invites():
|
|||||||
@bp.delete("/<invite_id>")
|
@bp.delete("/<invite_id>")
|
||||||
@require_admin
|
@require_admin
|
||||||
async def revoke_invite(invite_id: str):
|
async def revoke_invite(invite_id: str):
|
||||||
try:
|
iid = parse_uuid(invite_id)
|
||||||
iid = uuid.UUID(invite_id)
|
if iid is None:
|
||||||
except ValueError:
|
return not_found()
|
||||||
return jsonify({"error": "not found"}), 404
|
|
||||||
now = datetime.now(timezone.utc)
|
now = datetime.now(timezone.utc)
|
||||||
async with session_scope() as db:
|
async with session_scope() as db:
|
||||||
invite = await db.get(Invite, iid)
|
invite = await db.get(Invite, iid)
|
||||||
if invite is None:
|
if invite is None:
|
||||||
return jsonify({"error": "not found"}), 404
|
return not_found()
|
||||||
# A used invite has done its work and its record stays as it is; an already
|
# A used invite has done its work and its record stays as it is; an already
|
||||||
# revoked one keeps its first revocation time.
|
# revoked one keeps its first revocation time.
|
||||||
if invite.redeemed_at is None and invite.revoked_at is None:
|
if invite.redeemed_at is None and invite.revoked_at is None:
|
||||||
|
|||||||
@@ -64,9 +64,7 @@ async def test_email():
|
|||||||
async with session_scope() as db:
|
async with session_scope() as db:
|
||||||
cfg = await mail_settings(db)
|
cfg = await mail_settings(db)
|
||||||
if cfg is None:
|
if cfg is None:
|
||||||
return jsonify(
|
return json_error("Email is off: it needs an SMTP server, a from address and the public address.", 400)
|
||||||
{"error": "Email is off: it needs an SMTP server, a from address and the public address."}
|
|
||||||
), 400
|
|
||||||
admin = await db.get(User, g.user_id)
|
admin = await db.get(User, g.user_id)
|
||||||
site = await get_setting(db, "site_name")
|
site = await get_setting(db, "site_name")
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user