feat(api): migrate default API version from v2 to v3

README.md

@@ -21,6 +21,12 @@ Set `TRUSTIFY_URL` to point to your Trustify instance:
 export TRUSTIFY_URL="https://trustify.example.com"
 ```
 
+When the URL has no API path, `/api/v2/` is appended. To use another API version, include the full path:
+
+```bash
+export TRUSTIFY_URL="https://atlas.release.devshift.net/api/v3/"
+```
+
 ### Authentication (Optional)
 
 If your Trustify instance requires authentication, also set `AUTH_ENDPOINT`:

src/trustshell/__init__.py

@@ -1,22 +1,21 @@
-import time
 import importlib.metadata
 import logging
 import os
 import sys
-from urllib.parse import urlparse, urlunparse, quote, parse_qs, urlencode
-from typing import Optional, Any
+import time
+import webbrowser
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from typing import Any
+from urllib.parse import parse_qs, quote, urlencode, urlparse, urlunparse
 
 import httpx
 import jwt
+from anytree import Node, RenderTree
 from packageurl import PackageURL
 from rich.console import Console
 from rich.logging import RichHandler
 from rich.theme import Theme
-import webbrowser
 from univers.versions import RpmVersion
-from anytree import Node, RenderTree
-
-from http.server import BaseHTTPRequestHandler, HTTPServer
 
 from trustshell.oidc.oidc_pkce_authcode import (
     LOCAL_SERVER_PORT,
@@ -42,13 +41,12 @@
 if "TRUSTIFY_URL" in os.environ:
     url_env = os.getenv("TRUSTIFY_URL", "")
     parsed_url = urlparse(url_env)
-    if not parsed_url.path or parsed_url.path != TRUSTIFY_URL_PATH:
+    if not parsed_url.path or parsed_url.path == "/":
         TRUSTIFY_URL = urlunparse(
             (parsed_url.scheme, parsed_url.netloc, TRUSTIFY_URL_PATH, "", "", "")
         )
     else:
         TRUSTIFY_URL = url_env
-    # Only enable authentication if AUTH_ENDPOINT is also set
     AUTH_ENABLED = bool(os.getenv("AUTH_ENDPOINT"))
 else:
     TRUSTIFY_URL = "http://localhost:8080/api/v2/"
@@ -105,9 +103,7 @@ def get_tag_from_purl(purl: PackageURL) -> str:
     return tag
 
 
-def build_node_purl(
-    purls: list[str], show_versions: bool = False
-) -> Optional[PackageURL]:
+def build_node_purl(purls: list[str], show_versions: bool = False) -> PackageURL | None:
     """
     Generate a base purl with a version or tag qualifier from a list of purls with homogenous
     type/namespace, and name
@@ -220,7 +216,7 @@ def check_or_get_access_token() -> str:
         console.print(
             "Unable to authenticate to Atlas, please try again after authenticating in the browser."
         )
-        exit(0)
+        sys.exit(0)
     return access_token
 
 
@@ -369,52 +365,70 @@ def make_request_with_retry(
         first_response = make_request_with_retry(client, query_params, auth_header)
         first_result = first_response.json()
 
-        total_available = first_result.get("total", 0)
-        if total_available == 0:
+        all_items = first_result.get("items", [])
+        total_available = first_result.get("total")
+        total_known = total_available is not None
+
+        if not all_items and (not total_known or total_available == 0):
             if component_name:
                 console.print(f"No items found for {component_name}")
             return {"items": [], "total": 0}
 
-        all_items = first_result.get("items", [])
-        total_pages = (total_available + limit - 1) // limit
-
-        if logger.isEnabledFor(logging.DEBUG):
-            logger.debug(
-                f"Paginated request: {total_available} total items, "
-                f"{total_pages} page(s), page 1/{total_pages} complete"
-            )
+        if total_known:
+            total_pages = (total_available + limit - 1) // limit
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug(
+                    f"Paginated request: {total_available} total items, "
+                    f"{total_pages} page(s), page 1/{total_pages} complete"
+                )
+        else:
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug(
+                    f"Paginated request: total unknown, "
+                    f"page 1 returned {len(all_items)} items"
+                )
 
-        # Fetch remaining pages sequentially
         offset = limit
         page_num = 2
-        while offset < total_available:
+        while True:
+            if total_known and offset >= total_available:
+                break
+            if not total_known and len(all_items) - (offset - limit) < limit:
+                break
+
             page_params = {**base_params, "limit": limit, "offset": offset}
             if logger.isEnabledFor(logging.DEBUG):
-                logger.debug(
-                    f"Fetching page {page_num}/{total_pages} (offset {offset})..."
-                )
+                if total_known:
+                    total_pages = (total_available + limit - 1) // limit
+                    logger.debug(
+                        f"Fetching page {page_num}/{total_pages} (offset {offset})..."
+                    )
+                else:
+                    logger.debug(f"Fetching page {page_num} (offset {offset})...")
             try:
                 response = make_request_with_retry(client, page_params, auth_header)
                 result = response.json()
                 page_items = result.get("items", [])
                 all_items.extend(page_items)
                 if logger.isEnabledFor(logging.DEBUG):
                     logger.debug(
-                        f"Page {page_num}/{total_pages} complete "
-                        f"({len(all_items)}/{total_available} items)"
+                        f"Page {page_num} complete ({len(all_items)} items so far)"
                     )
+                if not page_items or len(page_items) < limit:
+                    break
                 offset += limit
                 page_num += 1
-            except Exception as e:
+            except httpx.HTTPError as e:
                 logger.error(f"Error fetching page at offset {offset}: {e}")
                 break
 
+        total_count = total_available if total_known else len(all_items)
         if component_name:
             console.print(
-                f"Retrieved {len(all_items)} items out of {total_available} total for {component_name}"
+                f"Retrieved {len(all_items)} items out of {total_count} total for {component_name}"
             )
 
-        return {"items": all_items, "total": total_available}
+        return {"items": all_items, "total": total_count}
 
 
 def render_tree_to_string(root: Node) -> str:

src/trustshell/api.py

@@ -1,9 +1,9 @@
-import click
-import httpx
 import json
 import logging
 from urllib.parse import quote
 
+import click
+import httpx
 from rich.console import Console
 from rich.theme import Theme
 
@@ -78,9 +78,7 @@ def api(endpoint: str, subpath: str, params: tuple[str], debug: bool) -> None:
             f"HTTP error {exc.response.status_code}: {exc.response.text}", style="error"
         )
     except httpx.RequestError as exc:
-        console.print(f"Request error: {str(exc)}", style="error")
+        console.print(f"Request error: {exc!s}", style="error")
     except json.JSONDecodeError as exc:
         console.print("Response is not valid JSON:", style="warning")
-        console.print(f"JSON decode error: {str(exc)}")
-    except Exception as exc:
-        console.print(f"Unexpected error: {str(exc)}", style="error")
+        console.print(f"JSON decode error: {exc!s}")

src/trustshell/models.py

@@ -1,7 +1,6 @@
 """Data models for trust-products search results."""
 
 from dataclasses import dataclass, field
-from typing import Optional
 
 
 @dataclass(frozen=True)
@@ -18,7 +17,7 @@ class ProductResultRow:
 
     cpe: str
     ps_update_stream: str
-    ps_module: Optional[str]
+    ps_module: str | None
     matched_component: str  # PURL that matched (important for wildcard search)
     shipped_component: (
         str  # PURL for affects: image-index/arch-specific OCI, or SRPM/binary RPM

src/trustshell/oidc/__init__.py

@@ -4,23 +4,23 @@
 """
 
 from .oidc_pkce_authcode import (
-    gen_things,
+    AUTH_ENDPOINT,
+    LOCAL_SERVER_PORT,
+    REDIRECT_URI,
     build_url,
     code_to_token,
-    get_fresh_token,
+    gen_things,
     get_client_credentials_token,
-    LOCAL_SERVER_PORT,
-    REDIRECT_URI,
-    AUTH_ENDPOINT,
+    get_fresh_token,
 )
 
 __all__ = [
-    "gen_things",
+    "AUTH_ENDPOINT",
+    "LOCAL_SERVER_PORT",
+    "REDIRECT_URI",
     "build_url",
     "code_to_token",
-    "get_fresh_token",
+    "gen_things",
     "get_client_credentials_token",
-    "LOCAL_SERVER_PORT",
-    "REDIRECT_URI",
-    "AUTH_ENDPOINT",
+    "get_fresh_token",
 ]

src/trustshell/oidc/oidc_pkce_authcode.py

@@ -1,15 +1,12 @@
-#!/usr/bin/env python
-
 import json
 import logging
 import os
 import secrets
-
 import urllib.parse
 
+import httpx
 import jwt
 import pkce
-import httpx
 
 logger = logging.getLogger("trustshell")
 # Client ID - Keycloak uses "atlas-frontend"; Cognito TPA uses a dedicated CLI client
@@ -61,7 +58,7 @@ def get_client_credentials_token() -> str | None:
 
 
 def gen_things() -> tuple[str, str, str]:
-    logging.debug("Generating verifier, challenge, state")
+    logger.debug("Generating verifier, challenge, state")
     code_verifier, code_challenge = pkce.generate_pkce_pair()
     state = secrets.token_urlsafe(16)
     logger.debug(f"Code Verifier: {code_verifier}")
@@ -155,6 +152,5 @@ def get_fresh_token(refresh_token: str) -> tuple[str, str]:
     logger.debug("Each raw part of the response body:")
     for k, v in r2a_json.items():
         logger.debug(f"{k}:{v}")
-    else:
-        logger.debug(f"Access Token: {access_token}")
+    logger.debug(f"Access Token: {access_token}")
     return access_token, refresh_token

src/trustshell/oidc/oidc_pkce_server.py

@@ -2,8 +2,8 @@
 
 import json
 import os
-from http.server import SimpleHTTPRequestHandler
 import socketserver
+from http.server import SimpleHTTPRequestHandler
 from typing import Any
 from urllib.parse import (
     parse_qs,
@@ -13,18 +13,18 @@
 try:
     # When run as part of the trustshell package
     from .oidc_pkce_authcode import (
+        AUTH_ENDPOINT,
         code_to_token,
         gen_things,
         get_fresh_token,
-        AUTH_ENDPOINT,
     )
 except ImportError:
     # When run as a standalone script
     from oidc_pkce_authcode import (  # type: ignore[import-not-found,no-redef]
+        AUTH_ENDPOINT,
         code_to_token,
         gen_things,
         get_fresh_token,
-        AUTH_ENDPOINT,
     )
 
 PORT: int = int(os.getenv("LISTEN_PORT", "8080"))

src/trustshell/osidb.py

@@ -1,18 +1,18 @@
-from collections import defaultdict
 import logging
 import os
 import subprocess
 import sys
 import tempfile
-from typing import Any, Union
+from collections import defaultdict
+from typing import Any
 
 import click
-
-from trustshell.models import Affect
-from requests import HTTPError
-from trustshell import console
 import osidb_bindings
 from osidb_bindings.bindings.python_client.models import Flaw
+from requests import HTTPError
+
+from trustshell import console
+from trustshell.models import Affect
 
 logger = logging.getLogger(__name__)
 
@@ -21,9 +21,7 @@ class OSIDB:
     def __init__(self) -> None:
         endpoint = os.getenv("OSIDB_ENDPOINT")
         if endpoint is None:
-            raise EnvironmentError(
-                "The environment variable 'OSIDB_ENDPOINT' is not set."
-            )
+            raise OSError("The environment variable 'OSIDB_ENDPOINT' is not set.")
         self.session = osidb_bindings.new_session(osidb_server_uri=endpoint)  # type: ignore[attr-defined]
 
     @staticmethod
@@ -52,7 +50,7 @@ def parse_stream_purl_tuples(tuples_list: list[str]) -> set[tuple[str, str]]:
 
     @staticmethod
     def edit_tuples_in_editor(
-        current_tuples: Union[list[tuple[str, str]], set[tuple[str, str]]],
+        current_tuples: list[tuple[str, str]] | set[tuple[str, str]],
     ) -> list[tuple[str, str]]:
         """
         Opens the default text editor for the user to modify the ps_update_stream/purl tuples.
@@ -78,13 +76,13 @@ def edit_tuples_in_editor(
                 f"Error: Editor '{editor}' not found. Please set your EDITOR environment variable.",
                 style="error",
             )
-            exit(1)
+            sys.exit(1)
         except subprocess.CalledProcessError:
             console.print(
                 "Editor exited with an error. Changes might not be saved.",
                 style="error",
             )
-            exit(1)
+            sys.exit(1)
 
         with open(temp_filepath, "r") as file:
             modified_content = file.read()
@@ -121,7 +119,7 @@ def add_affects(self, flaw: Flaw, affects_to_add: list[Affect]) -> None:
         except HTTPError as e:
             msg = e.response.text
             console.print(f"Failed to update flaw: {e}: {msg}")
-            exit(1)
+            sys.exit(1)
         console.print(f"Added {len(bulk_create_response.results)} new affects")
 
     def edit_flaw_affects(
@@ -139,7 +137,7 @@ def edit_flaw_affects(
 
         try:
             flaw = self.session.flaws.retrieve(id=flaw_id)
-        except Exception as e:
+        except (OSError, RuntimeError, ValueError, TypeError, KeyError) as e:
             console.print(f"Could not retrieve flaw {flaw_id}: {e}")
             return
 
@@ -251,7 +249,7 @@ def edit_flaw_affects(
                             f"Failed to delete flaw affect {existing_key}: {e}: {msg}",
                             style="error",
                         )
-                        exit(1)
+                        sys.exit(1)
 
             # Add any new affects not already on the flaw in NEW state
             ps_stream_purls_as_affects = [