centralizing error reporting

Author: Redbot

src/redfetch/api.py

@@ -30,7 +30,7 @@ async def get_api_headers():
         if not user_id:
             user_id = await fetch_user_id_from_api(api_key)
             if not user_id:
-                raise Exception("Unable to retrieve user ID using the provided API key.")
+                raise RuntimeError("Unable to retrieve user ID using the provided API key.")
         headers['XF-Api-User'] = str(user_id)
         return headers
 
@@ -50,10 +50,7 @@ async def get_api_headers():
             redirect_uri = os.environ.get("REDFETCH_OAUTH_REDIRECT_URI") or (settings.get("OAUTH_REDIRECT_URI") if settings else None) or "http://127.0.0.1:62897/"
 
             if not client_id:
-                raise Exception(
-                    "OAuth refresh token is present but OAUTH_CLIENT_ID is not configured. "
-                    "Set REDFETCH_OAUTH_CLIENT_ID (or OAUTH_CLIENT_ID in settings.local.toml) and try again."
-                )
+                raise RuntimeError("OAuth client is not configured.")
 
             refreshed = await asyncio.to_thread(
                 auth.refresh_token,
@@ -66,12 +63,12 @@ async def get_api_headers():
                 if access_token:
                     return {"Authorization": f"Bearer {access_token}"}
 
-            raise Exception("OAuth token refresh failed. Please run `redfetch logout` and authorize again.")
+            raise RuntimeError("OAuth token refresh failed. Please run `redfetch logout` and authorize again.")
 
         # Access token exists but is expired/missing expiry and there's no refresh token available.
-        raise Exception("OAuth access token is expired and no refresh token is available. Please authorize again.")
+        raise RuntimeError("OAuth access token is expired and no refresh token is available. Please authorize again.")
 
-    raise Exception(
+    raise RuntimeError(
         "Not authenticated. Set REDGUIDES_API_KEY (and optionally REDGUIDES_USER_ID), "
         "or authorize via OAuth."
     )
@@ -336,7 +333,7 @@ async def get_username():
         username = await fetch_username(api_key)
         if username != "Unknown":
             return username
-        raise Exception("Unable to retrieve username using the provided API key.")
+        raise RuntimeError("Unable to retrieve username using the provided API key.")
 
     # Prefer OAuth bearer token if present (including refresh-only sessions)
     access_token = keyring.get_password(KEYRING_SERVICE_NAME, "access_token")
@@ -349,6 +346,6 @@ async def get_username():
             set_username(me["username"])
             set_user_id(me["user_id"])
             return me["username"]
-        raise Exception("Unable to retrieve username using the stored OAuth token.")
+        raise RuntimeError("Unable to retrieve username using the stored OAuth token.")
 
-    raise Exception("Username not found. Set REDGUIDES_API_KEY or authorize via OAuth.")
+    raise RuntimeError("Username not found. Set REDGUIDES_API_KEY or authorize via OAuth.")

src/redfetch/auth.py

@@ -9,7 +9,6 @@
 import base64
 import hashlib
 import os
-import sys
 import time
 import webbrowser
 from datetime import datetime, timedelta
@@ -91,7 +90,7 @@ def do_GET(self):
         self.wfile.write(b"Authorization successful. You can close this tab.")
 
 
-def first_authorization(client_id: str, client_secret: str | None, *, scope: str, redirect_uri: str) -> bool:
+def first_authorization(client_id: str, client_secret: str | None, *, scope: str, redirect_uri: str) -> None:
     """Perform auth via browser and cache tokens.
 
     Uses Authorization Code + PKCE (S256) as required by XF for public clients.
@@ -145,14 +144,16 @@ def first_authorization(client_id: str, client_secret: str | None, *, scope: str
     headers = {"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}
     response = httpx.post(TOKEN_ENDPOINT, headers=headers, data=payload, timeout=10.0)
     if not response.is_success:
-        print("Failed to retrieve tokens.")
-        print(response.text)
-        return False
+        details = response.text.strip()
+        if details:
+            raise RuntimeError(f"Failed to retrieve tokens.\n{details}")
+        raise RuntimeError("Failed to retrieve tokens.")
 
     token_data = response.json()
     if token_data.get("error"):
-        print(f"OAuth token error: {token_data.get('error')} {token_data.get('error_description', '')}".strip())
-        return False
+        raise RuntimeError(
+            f"OAuth token error: {token_data.get('error')} {token_data.get('error_description', '')}".strip()
+        )
 
     # Step 5: Cache tokens and basic user info
     store_tokens_in_keyring(token_data)
@@ -164,8 +165,6 @@ def first_authorization(client_id: str, client_secret: str | None, *, scope: str
     except Exception:
         pass
 
-    return True
-
 
 def _cache_user_info(access_token: str | None) -> None:
     """Fetch /api/me and cache username/user_id (best-effort)."""
@@ -195,11 +194,7 @@ def authorize():
     redirect_uri = _get_setting("OAUTH_REDIRECT_URI", DEFAULT_REDIRECT_URI)
 
     if not client_id:
-        print("OAuth client is not configured.")
-        print("Set one of the following:")
-        print("  - Environment variable: REDFETCH_OAUTH_CLIENT_ID")
-        print("  - Or add to your settings.local.toml: OAUTH_CLIENT_ID = \"...\"")
-        sys.exit(1)
+        raise RuntimeError("OAuth client is not configured.")
 
     data = get_cached_tokens()
 
@@ -216,9 +211,7 @@ def authorize():
 
     # Fall back to interactive authorization
     print("Performing full authorization...")
-    if not first_authorization(client_id, client_secret, scope=scope, redirect_uri=redirect_uri):
-        print("Authorization failed.")
-        sys.exit(1)
+    first_authorization(client_id, client_secret, scope=scope, redirect_uri=redirect_uri)
 
 
 def _port_from_redirect_uri(redirect_uri: str) -> int:
@@ -372,16 +365,18 @@ def initialize_keyring():
         # Attempt to use the keyring to trigger any potential errors
         keyring.get_password('test_service', 'test_user')
     except (NoKeyringError, ModuleNotFoundError):
-        print("No suitable keyring backend found, probably because you're not on Windows.")
-        print("Please install `keyrings.alt` by running:")
-        print("    pip install keyrings.alt")
-        print("Then restart the application.")
-        sys.exit(1)
+        raise RuntimeError(
+            "No suitable keyring backend found, probably because you're not on Windows.\n\n"
+            "Please install `keyrings.alt` by running:\n"
+            "    pip install keyrings.alt\n\n"
+            "Then restart the application."
+        )
     except Exception as e:
         # Catch any other exceptions that may occur and handle them gracefully
-        print(f"An error occurred while initializing keyring: {e}")
-        print("Please ensure that a suitable keyring backend is available.")
-        sys.exit(1)
+        raise RuntimeError(
+            f"An error occurred while initializing keyring: {e}\n\n"
+            "Please ensure that a suitable keyring backend is available."
+        ) from e
 
 
 if __name__ == "__main__":

src/redfetch/config_firstrun.py

@@ -1,7 +1,6 @@
 # Standard
 import os
 import platform
-import sys
 import json
 
 # Third-party
@@ -300,7 +299,7 @@ def first_run_setup():
     elif any(word in response for word in ["no", "nope", "nah", "nay"]) or response == "n":
         console.print("\n[bold red]The wizards point to the sky with their longest finger ... \"BEGONE!\"[/bold red]")
         Prompt.ask("\nPress Enter to continue")
-        sys.exit(1)
+        raise SystemExit(1)
     elif any(phrase in response for phrase in ["xyzzy", "plugh", "hello sailor", "mailbox", "east", "leave house", "grue"]):
         console.print(
             "\nAs you utter the ancient words, the wizards eyes widen."
@@ -318,7 +317,7 @@ def first_run_setup():
     else:
         console.print("\n[bold red]The wizards shake their heads sadly, \"Your riddle eludes us. Perhaps you should go east.\"[/bold red]")
         Prompt.ask("\nPress Enter to continue")
-        sys.exit(1)
+        raise SystemExit(1)
 
     config_dir = setup_directories()
     create_first_run_flag(default_config_dir, config_dir)

src/redfetch/main.py

@@ -22,6 +22,7 @@
 from redfetch import push
 from redfetch import sync as sync_pipeline
 from redfetch import store
+from redfetch.runtime_errors import exit_with_fatal_error
 
 
 app = typer.Typer(
@@ -640,11 +641,18 @@ def root(
 # ============================================================================
 
 def main():
-    # Launch TUI when no arguments are provided
-    if len(sys.argv) == 1:
-        run_tui()
-        return
-    app()
+    try:
+        # Launch TUI when no arguments are provided
+        if len(sys.argv) == 1:
+            run_tui()
+            return
+        app()
+    except typer.Exit:
+        raise
+    except KeyboardInterrupt:
+        raise
+    except Exception as exc:
+        exit_with_fatal_error(exc)
 
 
 if __name__ == "__main__":

src/redfetch/push.py

@@ -2,12 +2,12 @@
 import asyncio
 import httpx
 import keepachangelog
+import typer
 from md2bbcode.main import process_readme
 
 from redfetch import api
 from redfetch.net import BASE_URL
 from redfetch import auth
-import sys
 
 XF_API_URL = f'{BASE_URL}/api'
 URI_MESSAGE = f'{XF_API_URL}/resource-updates'
@@ -220,32 +220,27 @@ def handle_cli(args):
 
     if not any([args.description, args.version, args.message, args.file]):
         print("At least one option (--description, --version, --message, or --file) must be specified.")
-        sys.exit(1)
+        raise typer.Exit(code=1)
 
     if args.message and not args.version:
         print("The --message option requires --version to be specified.")
-        sys.exit(1)
-
-    try:
-        # Ensure the user is authorized
-        auth.initialize_keyring()
-        auth.authorize()
-
-        # Blocking call is fine here; push is a short-lived CLI operation.
-        resource = asyncio.run(api.get_resource_details(args.resource_id))
-        resource_id = resource['resource_id']
-
-        if args.description:
-            update_description(resource_id, args.description, domain=args.domain)
-
-        if args.version and args.message:
-            message = generate_version_message(args)
-            version_info = {'version_string': args.version, 'message': message}
-            update_resource(resource_id, version_info, args.file)
-        elif args.file:
-            # Allow publishing a version with a file but no changelog message.
-            add_xf_attachment(resource_id, args.file, args.version)
-
-    except Exception as e:
-        print(f"An error occurred: {e}")
-        sys.exit(1)
+        raise typer.Exit(code=1)
+
+    # Ensure the user is authorized
+    auth.initialize_keyring()
+    auth.authorize()
+
+    # Blocking call is fine here; push is a short-lived CLI operation.
+    resource = asyncio.run(api.get_resource_details(args.resource_id))
+    resource_id = resource['resource_id']
+
+    if args.description:
+        update_description(resource_id, args.description, domain=args.domain)
+
+    if args.version and args.message:
+        message = generate_version_message(args)
+        version_info = {'version_string': args.version, 'message': message}
+        update_resource(resource_id, version_info, args.file)
+    elif args.file:
+        # Allow publishing a version with a file but no changelog message.
+        add_xf_attachment(resource_id, args.file, args.version)

src/redfetch/runtime_errors.py

@@ -0,0 +1,96 @@
+"""Shared fatal error helpers for startup/CLI boundaries."""
+
+from __future__ import annotations
+
+import os
+import sys
+import traceback
+from typing import NoReturn
+
+
+def is_windows_pyapp() -> bool:
+    """are we using the redfetch.exe executable on windows?"""
+    return sys.platform == "win32" and bool(os.getenv("PYAPP"))
+
+def normalize_error_message(
+    message: str | BaseException | None,
+    *,
+    fallback: str = "An unexpected error occurred.",
+) -> str:
+    """Make a safe non-empty string."""
+    if isinstance(message, BaseException):
+        normalized = str(message).strip()
+        if normalized:
+            return normalized
+        return f"{message.__class__.__name__} ({fallback})"
+
+    if message is None:
+        return fallback
+
+    normalized = str(message).strip()
+    if normalized:
+        return normalized
+    return fallback
+
+
+def _show_windows_error_dialog(message: str) -> None:
+    """Shows a Windows MessageBox."""
+    try:
+        import ctypes
+
+        ctypes.windll.user32.MessageBoxW(0, message, "redfetch", 0x10)
+    except Exception:
+        # Best effort only: CLI stderr output remains the source of truth.
+        pass
+
+
+def _format_error_details(error: BaseException) -> str:
+    """Summarize the exception with traceback and details."""
+    summary = f"{error.__class__.__name__}: {normalize_error_message(error)}"
+    traceback_text = "".join(
+        traceback.format_exception(type(error), error, error.__traceback__)
+    ).strip()
+    if not traceback_text:
+        return summary
+    return f"{summary}\n\n{traceback_text}"
+
+
+def display_fatal_error(
+    message: str | BaseException | None,
+) -> str:
+    """Print a fatal error and optionally show a Windows MessageBox."""
+    normalized_message = normalize_error_message(message)
+    error_details = (
+        _format_error_details(message)
+        if isinstance(message, BaseException)
+        else None
+    )
+
+    if is_windows_pyapp():
+        if error_details:
+            dialog_message = (
+                "Tip: Press Ctrl+C to copy this error report.\n\n"
+                f"{error_details}"
+            )
+            print(error_details, file=sys.stderr)
+        else:
+            dialog_message = normalized_message
+            print(normalized_message, file=sys.stderr)
+        _show_windows_error_dialog(dialog_message)
+    else:
+        # CLI behavior: print the raw normalized message without extra wrappers.
+        if error_details:
+            print(error_details, file=sys.stderr)
+        else:
+            print(normalized_message, file=sys.stderr)
+
+    return normalized_message
+
+
+def exit_with_fatal_error(
+    message: str | BaseException | None,
+    exit_code: int = 1,
+) -> NoReturn:
+    """Display a fatal error and terminate with a non-zero exit code."""
+    display_fatal_error(message)
+    raise SystemExit(exit_code)