Refactor API interactions in CLI

capture_tests/expected_captures.txt

@@ -671,12 +671,12 @@ Stderr:
 
 ############################
 Command: cbrain         tag update 99 --name Renamed --user-id 2 --group-id 3
-Status: 1
-Stdout: 37 bytes
+Status: 0
+Stdout: 29 bytes
 Stderr: 0 bytes
 
 Stdout:
-Failed: Invalid response from server
+Tag 99 updated successfully!
 Stderr:
 (No output)
 

cbrain_cli/cli_utils.py

@@ -2,9 +2,11 @@
 import json
 import re
 import urllib.error
+import urllib.parse
+import urllib.request
 
 # import importlib.metadata
-from cbrain_cli.config import CREDENTIALS_FILE
+from cbrain_cli.config import CREDENTIALS_FILE, DEFAULT_HEADERS, auth_headers
 
 try:
     # MARK: Credentials.
@@ -23,6 +25,40 @@
     cbrain_timestamp = None
 
 
+class CliValidationError(Exception):
+    """Raised when command arguments fail client-side validation.
+
+    Parameters
+    ----------
+    message : str
+        Human-readable error description.
+    field : str, optional
+        The CLI flag or argument name that caused the error (e.g. ``--per-page``).
+    """
+
+    def __init__(self, message, field=None):
+        super().__init__(message)
+        self.field = field
+
+    def __str__(self):
+        message = self.args[0] if self.args else ""
+        if self.field:
+            return f"{message} ({self.field})"
+        return message
+
+
+class CliApiError(Exception):
+    """
+    Raised when the API returns an expected error response.
+    """
+
+
+class CliResponseError(Exception):
+    """
+    Raised when the API response is malformed or unexpected.
+    """
+
+
 def is_authenticated():
     """
     Check if the user is authenticated.
@@ -85,7 +121,7 @@ def handle_connection_error(error):
         if error.code == 401:
             print(f"{status_description}: {error.reason}")
             print("Error: Access denied. Please log in using authorized credentials.")
-        elif error.code == 404 or error.code == 422 or error.code == 500:
+        elif error.code in (400, 404, 422, 500):
             # Try to extract specific error message from response
             try:
                 # Check if the error response has already been read
@@ -107,6 +143,14 @@ def handle_connection_error(error):
                             or error_data.get("notice")
                             or str(error_data)
                         )
+                        # Check if this looks like a password change redirect
+                        if "change_password" in error_msg:
+                            print(
+                                f"{status_description}: Account requires "
+                                "a password change. "
+                                "Please log into the web portal."
+                            )
+                            return
                         print(f"{status_description}: {error_msg}")
                         return
                 except json.JSONDecodeError:
@@ -178,6 +222,9 @@ def wrapper(*args, **kwargs):
         except KeyboardInterrupt:
             print("\nOperation cancelled")
             return 1
+        except (CliValidationError, CliApiError, CliResponseError) as e:
+            print(f"Error: {e}")
+            return 1
         except Exception as e:
             print(f"Operation failed: {str(e)}")
             return 1
@@ -209,6 +256,64 @@ def version_info(args):
     #     return 1
 
 
+def api_get(url, token, params=None):
+    """
+    Execute an authenticated GET request and return parsed JSON.
+    """
+    if params:
+        url = f"{url}?{urllib.parse.urlencode(params)}"
+    req = urllib.request.Request(url, headers=auth_headers(token), method="GET")
+    with urllib.request.urlopen(req) as r:
+        return json.loads(r.read().decode())
+
+
+def api_post_form(url, form_data, headers=None):
+    """
+    POST form-urlencoded data (unauthenticated) and return parsed JSON.
+    """
+    headers = headers or DEFAULT_HEADERS
+    body = urllib.parse.urlencode(form_data).encode()
+    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
+    with urllib.request.urlopen(req) as r:
+        return json.loads(r.read().decode())
+
+
+def api_send(url, token, method="POST", payload=None):
+    """
+    Execute an authenticated POST/PUT/DELETE request and return (data, status).
+    """
+    headers = auth_headers(token)
+    body = None
+    if payload is not None:
+        headers["Content-Type"] = "application/json"
+        body = json.dumps(payload).encode()
+    req = urllib.request.Request(url, data=body, headers=headers, method=method)
+    with urllib.request.urlopen(req) as r:
+        raw = r.read().decode()
+        return (json.loads(raw) if raw.strip() else {}), r.status
+
+
+def output_json(args, data):
+    """
+    Print data as JSON or JSONL if requested. Returns True if output was handled.
+    """
+    if getattr(args, "json", False):
+        json_printer(data)
+        return True
+    if getattr(args, "jsonl", False):
+        jsonl_printer(data)
+        return True
+    return False
+
+
+def display_key_value_table(pairs):
+    """
+    Print a (key-value) two-column Field/Value table from a list of (field, value) tuples.
+    """
+    rows = [{"field": k, "value": v} for k, v in pairs]
+    dynamic_table_print(rows, ["field", "value"], ["Field", "Value"])
+
+
 def json_printer(data):
     """
     Print data in JSON format.
@@ -235,13 +340,11 @@ def pagination(args, query_params):
     """
     per_page = getattr(args, "per_page", 25)
     if per_page < 5 or per_page > 1000:
-        print("Error: per-page must be between 5 and 1000")
-        return None
+        raise CliValidationError("per-page must be between 5 and 1000", field="--per-page")
 
     page = getattr(args, "page", 1)
     if page < 1:
-        print("Error: page must be 1 or greater")
-        return None
+        raise CliValidationError("page must be 1 or greater", field="--page")
 
     query_params["page"] = str(page)
     query_params["per_page"] = str(per_page)

cbrain_cli/data/background_activities.py

@@ -1,9 +1,4 @@
-import json
-import urllib.parse
-import urllib.request
-
-from cbrain_cli.cli_utils import api_token, cbrain_url
-from cbrain_cli.config import auth_headers
+from cbrain_cli.cli_utils import CliValidationError, api_get, api_token, cbrain_url
 
 
 def list_background_activities(args):
@@ -20,21 +15,7 @@ def list_background_activities(args):
     list or None
         List of background activity dictionaries if successful, None if error
     """
-    background_activities_endpoint = f"{cbrain_url}/background_activities"
-    headers = auth_headers(api_token)
-
-    request = urllib.request.Request(
-        background_activities_endpoint, data=None, headers=headers, method="GET"
-    )
-
-    try:
-        with urllib.request.urlopen(request) as response:
-            data = response.read().decode("utf-8")
-            background_activities_data = json.loads(data)
-            return background_activities_data
-    except Exception as e:
-        print(f"Error fetching background activities: {str(e)}")
-        return None
+    return api_get(f"{cbrain_url}/background_activities", api_token)
 
 
 def show_background_activity(args):
@@ -54,17 +35,5 @@ def show_background_activity(args):
     # Get the background activity ID from the --id argument
     activity_id = getattr(args, "id", None)
     if not activity_id:
-        print("Error: Background activity ID is required")
-        return None
-
-    background_activity_endpoint = f"{cbrain_url}/background_activities/{activity_id}"
-    headers = auth_headers(api_token)
-
-    request = urllib.request.Request(
-        background_activity_endpoint, data=None, headers=headers, method="GET"
-    )
-
-    with urllib.request.urlopen(request) as response:
-        data = response.read().decode("utf-8")
-        activity_data = json.loads(data)
-        return activity_data
+        raise CliValidationError("Background activity ID is required", field="id")
+    return api_get(f"{cbrain_url}/background_activities/{activity_id}", api_token)

cbrain_cli/data/data_providers.py

@@ -1,9 +1,12 @@
-import json
-import urllib.error
-import urllib.request
-
-from cbrain_cli.cli_utils import api_token, cbrain_url, pagination
-from cbrain_cli.config import auth_headers
+from cbrain_cli.cli_utils import (
+    CliApiError,
+    CliValidationError,
+    api_get,
+    api_send,
+    api_token,
+    cbrain_url,
+    pagination,
+)
 
 
 def show_data_provider(args):
@@ -22,27 +25,12 @@ def show_data_provider(args):
     """
     # Get the data provider ID from the --id argument.
     data_provider_id = getattr(args, "id", None)
-
     if not data_provider_id:
         return list_data_providers(args)
-
-    # Show specific data provider by ID
-    data_provider_endpoint = f"{cbrain_url}/data_providers/{data_provider_id}"
-    headers = auth_headers(api_token)
-
-    request = urllib.request.Request(
-        data_provider_endpoint, data=None, headers=headers, method="GET"
-    )
-
-    with urllib.request.urlopen(request) as response:
-        data = response.read().decode("utf-8")
-        provider_data = json.loads(data)
-
-    if provider_data.get("error"):
-        print(f"Error: {provider_data.get('error')}")
-        return None
-
-    return provider_data
+    data = api_get(f"{cbrain_url}/data_providers/{data_provider_id}", api_token)
+    if data.get("error"):
+        raise CliApiError(data.get("error"))
+    return data
 
 
 def list_data_providers(args):
@@ -59,23 +47,8 @@ def list_data_providers(args):
     list
         List of data provider dictionaries
     """
-    query_params = {}
-    query_params = pagination(args, query_params)
-
-    data_providers_endpoint = f"{cbrain_url}/data_providers"
-    query_string = urllib.parse.urlencode(query_params)
-    data_providers_endpoint = f"{data_providers_endpoint}?{query_string}"
-    headers = auth_headers(api_token)
-
-    request = urllib.request.Request(
-        data_providers_endpoint, data=None, headers=headers, method="GET"
-    )
-
-    with urllib.request.urlopen(request) as response:
-        data = response.read().decode("utf-8")
-        data_providers_data = json.loads(data)
-
-    return data_providers_data
+    params = pagination(args, {})
+    return api_get(f"{cbrain_url}/data_providers", api_token, params)
 
 
 def is_alive(args):
@@ -87,16 +60,10 @@ def is_alive(args):
     args : argparse.Namespace
         Command line arguments, including the id argument
     """
-    is_alive_endpoint = f"{cbrain_url}/data_providers/{args.id}/is_alive"
-    headers = auth_headers(api_token)
-
-    request = urllib.request.Request(is_alive_endpoint, data=None, headers=headers, method="GET")
-
-    with urllib.request.urlopen(request) as response:
-        data = response.read().decode("utf-8")
-        is_alive_data = json.loads(data)
-
-    return is_alive_data
+    data_provider_id = getattr(args, "id", None)
+    if not data_provider_id:
+        raise CliValidationError("Data provider ID is required", field="id")
+    return api_get(f"{cbrain_url}/data_providers/{data_provider_id}/is_alive", api_token)
 
 
 def delete_unregistered_files(args):
@@ -108,15 +75,10 @@ def delete_unregistered_files(args):
     args : argparse.Namespace
         Command line arguments, including the id argument
     """
-    delete_unregistered_files_endpoint = f"{cbrain_url}/data_providers/{args.id}/delete"
-    headers = auth_headers(api_token)
-
-    request = urllib.request.Request(
-        delete_unregistered_files_endpoint, data=None, headers=headers, method="POST"
+    data_provider_id = getattr(args, "id", None)
+    if not data_provider_id:
+        raise CliValidationError("Data provider ID is required", field="id")
+    data, _ = api_send(
+        f"{cbrain_url}/data_providers/{data_provider_id}/delete", api_token
     )
-
-    with urllib.request.urlopen(request) as response:
-        data = response.read().decode("utf-8")
-        delete_unregistered_files_data = json.loads(data)
-
-    return delete_unregistered_files_data
+    return data