Implement Phase 4 Authentication and API Requests in Cbrain-cli (Tasks 23–26)

cbrain_cli/cli_utils.py

@@ -10,13 +10,8 @@
 import urllib.request
 from pathlib import Path
 
-from cbrain_cli.config import DEFAULT_HEADERS, DEFAULT_TIMEOUT, auth_headers, load_credentials
-
-credentials = load_credentials() or {}
-cbrain_url = credentials.get("cbrain_url")
-api_token = credentials.get("api_token")
-user_id = credentials.get("user_id")
-cbrain_timestamp = credentials.get("timestamp")
+from cbrain_cli import config as cbrain_config
+from cbrain_cli.config import DEFAULT_HEADERS, DEFAULT_TIMEOUT, auth_headers
 
 PAGINATABLE_ACTIONS = {
     ("file", "list"),
@@ -62,11 +57,32 @@ class CliResponseError(Exception):
     """
 
 
+def get_auth():
+    """
+    Return (cbrain_url, api_token, user_id) from current credentials file.
+    """
+    creds = cbrain_config.load_credentials() or {}
+    return creds.get("cbrain_url"), creds.get("api_token"), creds.get("user_id")
+
+
+def _request_target(path, token=None):
+    """
+    Resolve (url, token) from path + optional token using call-time credentials.
+    """
+    if token is not None:
+        return path, token
+    base, token, _ = get_auth()
+    if path.startswith(("http://", "https://")):
+        return path, token
+    return f"{base}{path}", token
+
+
 def is_authenticated():
     """
     Check if the user is authenticated.
     """
     # Check if user is logged in.
+    cbrain_url, api_token, user_id = get_auth()
     if not api_token or not cbrain_url or not user_id:
         print("Not logged in. Use 'cbrain login' to login first.")
         return False
@@ -196,6 +212,7 @@ def handle_connection_error(error):
                 "Check your connection or set CBRAIN_TIMEOUT env var."
             )
         elif "Connection refused" in str(error):
+            cbrain_url, _, _ = get_auth()
             print(f"Error: Cannot connect to CBRAIN server at {cbrain_url}")
             print("Please check if the CBRAIN server is running and accessible.")
         else:
@@ -274,10 +291,11 @@ def version_info(args):
     return 0
 
 
-def api_get(url, token, params=None):
+def api_get(path, token=None, params=None):
     """
     Execute an authenticated GET request and return parsed JSON.
     """
+    url, token = _request_target(path, token)
     if params:
         url = f"{url}?{urllib.parse.urlencode(params)}"
     req = urllib.request.Request(url, headers=auth_headers(token), method="GET")
@@ -296,10 +314,12 @@ def api_post_form(url, form_data, headers=None):
         return json.loads(r.read().decode())
 
 
-def api_send(url, token, method="POST", payload=None):
+def api_send(path, token=None, method="POST", payload=None):
     """
-    Execute an authenticated POST/PUT/DELETE request and return (data, status).
+    Authenticated POST/PUT/DELETE. ``path`` is root-relative or absolute URL.
+    When ``token`` is omitted, credentials load at call time via ``get_auth()``.
     """
+    url, token = _request_target(path, token)
     headers = auth_headers(token)
     body = None
     if payload is not None:
@@ -311,6 +331,19 @@ def api_send(url, token, method="POST", payload=None):
         return (json.loads(raw) if raw.strip() else {}), r.status
 
 
+def api_post_multipart(path, body, content_type, token=None):
+    """
+    Authenticated multipart/form-data POST. Returns (parsed JSON, status).
+    """
+    url, token = _request_target(path, token)
+    headers = auth_headers(token)
+    headers["Content-Type"] = content_type
+    headers["Content-Length"] = str(len(body))
+    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=DEFAULT_TIMEOUT) as r:
+        return json.loads(r.read().decode()), r.status
+
+
 def output_json(args, data):
     """
     Print data as JSON or JSONL if requested. Returns True if output was handled.

cbrain_cli/data/background_activities.py

@@ -1,4 +1,4 @@
-from cbrain_cli.cli_utils import CliValidationError, api_get, api_token, cbrain_url
+from cbrain_cli.cli_utils import CliValidationError, api_get
 
 
 def list_background_activities(args):
@@ -15,7 +15,7 @@ def list_background_activities(args):
     list or None
         List of background activity dictionaries if successful, None if error
     """
-    return api_get(f"{cbrain_url}/background_activities", api_token)
+    return api_get("/background_activities")
 
 
 def show_background_activity(args):
@@ -36,4 +36,4 @@ def show_background_activity(args):
     activity_id = getattr(args, "id", None)
     if not activity_id:
         raise CliValidationError("Background activity ID is required", field="id")
-    return api_get(f"{cbrain_url}/background_activities/{activity_id}", api_token)
+    return api_get(f"/background_activities/{activity_id}")

cbrain_cli/data/data_providers.py

@@ -3,8 +3,6 @@
     CliValidationError,
     api_get,
     api_send,
-    api_token,
-    cbrain_url,
     pagination,
 )
 
@@ -27,7 +25,7 @@ def show_data_provider(args):
     data_provider_id = getattr(args, "id", None)
     if not data_provider_id:
         return list_data_providers(args)
-    data = api_get(f"{cbrain_url}/data_providers/{data_provider_id}", api_token)
+    data = api_get(f"/data_providers/{data_provider_id}")
     if data.get("error"):
         raise CliApiError(data.get("error"))
     return data
@@ -48,7 +46,7 @@ def list_data_providers(args):
         List of data provider dictionaries
     """
     params = pagination(args, {})
-    return api_get(f"{cbrain_url}/data_providers", api_token, params)
+    return api_get("/data_providers", params=params)
 
 
 def is_alive(args):
@@ -63,7 +61,7 @@ def is_alive(args):
     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)
+    return api_get(f"/data_providers/{data_provider_id}/is_alive")
 
 
 def delete_unregistered_files(args):
@@ -78,5 +76,5 @@ def delete_unregistered_files(args):
     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)
+    data, _ = api_send(f"/data_providers/{data_provider_id}/delete")
     return data

cbrain_cli/data/files.py

@@ -1,17 +1,13 @@
-import json
 import mimetypes
 import os
-import urllib.request
 
 from cbrain_cli.cli_utils import (
     CliValidationError,
     api_get,
+    api_post_multipart,
     api_send,
-    api_token,
-    cbrain_url,
     pagination,
 )
-from cbrain_cli.config import DEFAULT_TIMEOUT, auth_headers
 
 
 def show_file(args):
@@ -32,7 +28,7 @@ def show_file(args):
     file_id = getattr(args, "file", None)
     if not file_id:
         raise CliValidationError("File ID is required", field="file")
-    return api_get(f"{cbrain_url}/userfiles/{file_id}", api_token)
+    return api_get(f"/userfiles/{file_id}")
 
 
 def upload_file(args):
@@ -84,17 +80,10 @@ def upload_file(args):
         file_content = f.read()
 
     body = body_text.encode("utf-8") + file_content + f"\r\n--{boundary}--\r\n".encode()
-
-    headers = auth_headers(api_token)
-    headers["Content-Type"] = f"multipart/form-data; boundary={boundary}"
-    headers["Content-Length"] = str(len(body))
-
-    request = urllib.request.Request(
-        f"{cbrain_url}/userfiles", data=body, headers=headers, method="POST"
+    response_data, status = api_post_multipart(
+        "/userfiles", body, f"multipart/form-data; boundary={boundary}"
     )
-    with urllib.request.urlopen(request, timeout=DEFAULT_TIMEOUT) as response:
-        response_data = json.loads(response.read().decode("utf-8"))
-        return response_data, response.status, file_name, file_size, args.data_provider
+    return response_data, status, file_name, file_size, args.data_provider
 
 
 def _change_provider(args, operation):
@@ -111,7 +100,7 @@ def _change_provider(args, operation):
         "data_provider_id_for_mv_cp": dest_provider_id,
         operation: "",
     }
-    return api_send(f"{cbrain_url}/userfiles/change_provider", api_token, payload=payload)
+    return api_send("/userfiles/change_provider", payload=payload)
 
 
 def copy_file(args):
@@ -175,7 +164,7 @@ def list_files(args):
             params[key] = str(val)
 
     params = pagination(args, params)
-    return api_get(f"{cbrain_url}/userfiles", api_token, params)
+    return api_get("/userfiles", params=params)
 
 
 def delete_file(args):
@@ -196,8 +185,7 @@ def delete_file(args):
     if not file_id:
         raise CliValidationError("File ID is required", field="file_id")
     data, _ = api_send(
-        f"{cbrain_url}/userfiles/delete_files",
-        api_token,
+        "/userfiles/delete_files",
         method="DELETE",
         payload={"file_ids": [str(file_id)]},
     )

cbrain_cli/data/projects.py

@@ -5,8 +5,6 @@
     CliValidationError,
     api_get,
     api_send,
-    api_token,
-    cbrain_url,
 )
 from cbrain_cli.config import load_credentials, save_credentials
 
@@ -42,10 +40,10 @@ def switch_project(args):
             f"Invalid group ID '{group_id}'. Must be a number or 'all'", field="group_id"
         ) from None
 
-    _, switch_status = api_send(f"{cbrain_url}/groups/switch?id={group_id}", api_token)
+    _, switch_status = api_send(f"/groups/switch?id={group_id}")
     if switch_status not in (200, 201, 204):
         raise CliApiError(f"Failed to switch project (HTTP {switch_status})")
-    group_data = api_get(f"{cbrain_url}/groups/{group_id}", api_token)
+    group_data = api_get(f"/groups/{group_id}")
 
     credentials = load_credentials()
     if credentials is not None:
@@ -79,7 +77,7 @@ def unswitch_project(args):
         previous_group_name = credentials.get("current_group_name")
 
     if previous_group_id:
-        api_send(f"{cbrain_url}/groups/switch", api_token)
+        api_send("/groups/switch")
 
     if credentials is not None:
         credentials.pop("current_group_id", None)
@@ -114,7 +112,7 @@ def show_project(args):
     if project_id:
         # Show specific project by ID
         try:
-            return api_get(f"{cbrain_url}/groups/{project_id}", api_token)
+            return api_get(f"/groups/{project_id}")
         except urllib.error.HTTPError as e:
             if e.code == 404:
                 raise CliApiError(f"Project with ID {project_id} not found") from None
@@ -129,7 +127,7 @@ def show_project(args):
         return None
 
     try:
-        return api_get(f"{cbrain_url}/groups/{current_group_id}", api_token)
+        return api_get(f"/groups/{current_group_id}")
     except urllib.error.HTTPError as e:
         if e.code == 404:
             credentials.pop("current_group_id", None)
@@ -153,4 +151,4 @@ def list_projects(args):
     list
         List of project dictionaries
     """
-    return api_get(f"{cbrain_url}/groups", api_token)
+    return api_get("/groups")

cbrain_cli/data/remote_resources.py

@@ -1,4 +1,4 @@
-from cbrain_cli.cli_utils import CliValidationError, api_get, api_token, cbrain_url
+from cbrain_cli.cli_utils import CliValidationError, api_get
 
 
 def list_remote_resources(args):
@@ -15,7 +15,7 @@ def list_remote_resources(args):
     list
         List of remote resource dictionaries
     """
-    return api_get(f"{cbrain_url}/bourreaux", api_token)
+    return api_get("/bourreaux")
 
 
 def show_remote_resource(args):
@@ -35,4 +35,4 @@ def show_remote_resource(args):
     resource_id = getattr(args, "remote_resource", None)
     if not resource_id:
         raise CliValidationError("Remote resource ID is required", field="remote_resource")
-    return api_get(f"{cbrain_url}/bourreaux/{resource_id}", api_token)
+    return api_get(f"/bourreaux/{resource_id}")

cbrain_cli/data/tags.py

@@ -2,8 +2,6 @@
     CliValidationError,
     api_get,
     api_send,
-    api_token,
-    cbrain_url,
     pagination,
 )
 
@@ -36,7 +34,7 @@ def list_tags(args):
         List of tag dictionaries
     """
     params = pagination(args, {})
-    return api_get(f"{cbrain_url}/tags", api_token, params)
+    return api_get("/tags", params=params)
 
 
 def show_tag(args):
@@ -57,7 +55,7 @@ def show_tag(args):
     tag_id = getattr(args, "id", None)
     if not tag_id:
         raise CliValidationError("Tag ID is required", field="id")
-    return api_get(f"{cbrain_url}/tags/{tag_id}", api_token)
+    return api_get(f"/tags/{tag_id}")
 
 
 def create_tag(args):
@@ -76,7 +74,7 @@ def create_tag(args):
     """
     # Get tag details from command line arguments
     payload = _tag_payload(args)
-    data, status = api_send(f"{cbrain_url}/tags", api_token, payload=payload)
+    data, status = api_send("/tags", payload=payload)
     success = status in (200, 201, 204)
     return data, success, None, status
 
@@ -100,7 +98,7 @@ def update_tag(args):
     if not tag_id:
         raise CliValidationError("Tag ID is required", field="tag_id")
     payload = _tag_payload(args)
-    data, status = api_send(f"{cbrain_url}/tags/{tag_id}", api_token, method="PUT", payload=payload)
+    data, status = api_send(f"/tags/{tag_id}", method="PUT", payload=payload)
     success = status in (200, 201, 204)
     return data, success, None, status
 
@@ -123,6 +121,6 @@ def delete_tag(args):
     tag_id = getattr(args, "tag_id", None)
     if not tag_id:
         raise CliValidationError("Tag ID is required", field="tag_id")
-    _, status = api_send(f"{cbrain_url}/tags/{tag_id}", api_token, method="DELETE")
+    _, status = api_send(f"/tags/{tag_id}", method="DELETE")
     success = status in (200, 201, 204)
     return success, None, status