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.
@@ -85,7 +87,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 +109,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:
@@ -209,6 +219,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.

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 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):
@@ -56,15 +37,4 @@ def show_background_activity(args):
     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
+    return api_get(f"{cbrain_url}/background_activities/{activity_id}", api_token)

cbrain_cli/data/data_providers.py

@@ -1,9 +1,4 @@
-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 api_get, api_send, api_token, cbrain_url, pagination
 
 
 def show_data_provider(args):
@@ -22,27 +17,13 @@ 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')}")
+    data = api_get(f"{cbrain_url}/data_providers/{data_provider_id}", api_token)
+    if data.get("error"):
+        print(f"Error: {data.get('error')}")
         return None
-
-    return provider_data
+    return data
 
 
 def list_data_providers(args):
@@ -59,23 +40,10 @@ 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, {})
+    if params is None:
+        return None
+    return api_get(f"{cbrain_url}/data_providers", api_token, params)
 
 
 def is_alive(args):
@@ -87,16 +55,7 @@ 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
+    return api_get(f"{cbrain_url}/data_providers/{args.id}/is_alive", api_token)
 
 
 def delete_unregistered_files(args):
@@ -108,15 +67,5 @@ 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"
-    )
-
-    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
+    data, _ = api_send(f"{cbrain_url}/data_providers/{args.id}/delete", api_token)
+    return data