Enhance error handling and refactor API request logic

cbrain_cli/cli_utils.py

@@ -1,6 +1,6 @@
-import datetime
 import functools
 import json
+import re
 import urllib.error
 
 # import importlib.metadata
@@ -27,20 +27,44 @@ def is_authenticated():
     """
     Check if the user is authenticated.
     """
-
-    if cbrain_timestamp:
-        timestamp_obj = datetime.datetime.fromisoformat(cbrain_timestamp)
-        if datetime.datetime.now() - timestamp_obj > datetime.timedelta(days=1):
-            print("Session expired. Please log in again using 'cbrain login'.")
-            CREDENTIALS_FILE.unlink()
-            return False
     # Check if user is logged in.
     if not api_token or not cbrain_url or not user_id:
         print("Not logged in. Use 'cbrain login' to login first.")
         return False
     return True
 
 
+def get_status_code_description(status_code):
+    """
+    Get a user-friendly description for HTTP status codes.
+
+    Parameters
+    ----------
+    status_code : int
+        HTTP status code
+
+    Returns
+    -------
+    str
+        Description of the status code category
+    """
+    if 400 <= status_code < 500:
+        if status_code == 401:
+            return "Authentication error (401)"
+        elif status_code == 403:
+            return "Access forbidden (403)"
+        elif status_code == 404:
+            return "Resource not found (404)"
+        elif status_code == 422:
+            return "Validation error (422)"
+        else:
+            return f"Client error ({status_code})"
+    elif 500 <= status_code < 600:
+        return f"Server error ({status_code})"
+    else:
+        return f"HTTP error ({status_code})"
+
+
 def handle_connection_error(error):
     """
     Handle connection errors with informative messages including server URL.
@@ -55,20 +79,75 @@ def handle_connection_error(error):
     None
         Prints appropriate error messages
     """
-    if isinstance(error, urllib.error.URLError):
+    if isinstance(error, urllib.error.HTTPError):
+        status_description = get_status_code_description(error.code)
+
+        if error.code == 401:
+            print(f"{status_description}: {error.reason}")
+            print("Try with Authorized Access")
+        elif error.code == 404 or error.code == 422 or error.code == 500:
+            # Try to extract specific error message from response
+            try:
+                # Check if the error response has already been read
+                if hasattr(error, "read"):
+                    error_response = error.read().decode("utf-8")
+                elif hasattr(error, "response") and hasattr(error.response, "read"):
+                    error_response = error.response.read().decode("utf-8")
+                else:
+                    error_response = ""
+
+                # Try to parse as JSON first (for 422 validation errors)
+                try:
+                    error_data = json.loads(error_response)
+                    if isinstance(error_data, dict):
+                        # Look for common error message fields
+                        error_msg = (
+                            error_data.get("message")
+                            or error_data.get("error")
+                            or error_data.get("notice")
+                            or str(error_data)
+                        )
+                        print(f"{status_description}: {error_msg}")
+                        return
+                except json.JSONDecodeError:
+                    # Not JSON, try HTML parsing
+                    pass
+
+                # Extract header h1 and h2 content from HTML
+                header_pattern = r"<header>\s*<h1>(.*?)</h1>\s*</header>"
+                header_match = re.search(header_pattern, error_response, re.DOTALL)
+                h2_match = re.search(r"<h2>(.*?)</h2>", error_response)
+
+                error_parts = []
+
+                if header_match:
+                    header_text = header_match.group(1).strip()
+                    # Decode HTML entities
+                    header_text = header_text.replace("&#39;", "'").replace("&quot;", '"')
+                    error_parts.append(header_text)
+
+                if h2_match:
+                    h2_text = h2_match.group(1).strip()
+                    # Decode HTML entities
+                    h2_text = h2_text.replace("&#39;", "'").replace("&quot;", '"')
+                    # Remove SQL WHERE clause details
+                    h2_text = re.sub(r"\s*\[WHERE.*?\]", "", h2_text)
+                    error_parts.append(h2_text)
+
+                if error_parts:
+                    print(f"{status_description}: " + " - ".join(error_parts))
+                else:
+                    print(f"{status_description}: {error.reason}")
+            except Exception:
+                print(f"{status_description}: {error.reason}")
+        else:
+            print(f"{status_description}: {error.reason}")
+    elif isinstance(error, urllib.error.URLError):
         if "Connection refused" in str(error):
             print(f"Error: Cannot connect to CBRAIN server at {cbrain_url}")
             print("Please check if the CBRAIN server is running and accessible.")
         else:
             print(f"Connection failed: {error.reason}")
-    elif isinstance(error, urllib.error.HTTPError):
-        print(f"Request failed: HTTP {error.code} - {error.reason}")
-        if error.code == 401:
-            print("Invalid username or password")
-        elif error.code == 404:
-            print("Resource not found")
-        elif error.code == 500:
-            print("Internal server error")
     else:
         print(f"Connection error: {str(error)}")
 

cbrain_cli/data/background_activities.py

@@ -64,18 +64,7 @@ def show_background_activity(args):
         background_activity_endpoint, data=None, headers=headers, method="GET"
     )
 
-    try:
-        with urllib.request.urlopen(request) as response:
-            data = response.read().decode("utf-8")
-            activity_data = json.loads(data)
-            return activity_data
-
-    except urllib.error.HTTPError as e:
-        if e.code == 404:
-            print(f"Error: Background activity with ID {activity_id} not found")
-        else:
-            print(f"Error: HTTP {e.code} - {e.reason}")
-        return None
-    except Exception as e:
-        print(f"Error getting background activity details: {str(e)}")
-        return None
+    with urllib.request.urlopen(request) as response:
+        data = response.read().decode("utf-8")
+        activity_data = json.loads(data)
+        return activity_data

cbrain_cli/data/data_providers.py

@@ -34,24 +34,16 @@ def show_data_provider(args):
         data_provider_endpoint, data=None, headers=headers, method="GET"
     )
 
-    try:
-        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
-
-    except urllib.error.HTTPError as e:
-        if e.code == 404:
-            print(f"Error: Data provider with ID {data_provider_id} not found")
-        else:
-            print(f"Error: HTTP {e.code} - {e.reason}")
+    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
+
 
 def list_data_providers(args):
     """

cbrain_cli/data/files.py

@@ -48,7 +48,7 @@ def upload_file(args):
     Parameters
     ----------
     args : argparse.Namespace
-        Command line arguments including data_provider, file_type, file_path, and group_id
+        Command line arguments including data_provider, file_path, and group_id
 
     Returns
     -------
@@ -94,12 +94,6 @@ def upload_file(args):
     body_parts.append("")
     body_parts.append(str(group_id))
 
-    # Add file_type field.
-    body_parts.append(f"--{boundary}")
-    body_parts.append('Content-Disposition: form-data; name="file_type"')
-    body_parts.append("")
-    body_parts.append(args.file_type)
-
     # Add file data.
     body_parts.append(f"--{boundary}")
     body_parts.append(f'Content-Disposition: form-data; name="upload_file"; filename="{file_name}"')
@@ -127,21 +121,10 @@ def upload_file(args):
     upload_endpoint = f"{cbrain_url}/userfiles"
     request = urllib.request.Request(upload_endpoint, data=body, headers=headers, method="POST")
 
-    try:
-        with urllib.request.urlopen(request) as response:
-            data = response.read().decode("utf-8")
-            response_data = json.loads(data)
-            return response_data, response.status, file_name, file_size, args.data_provider
-
-    except urllib.error.HTTPError as e:
-        # Handle HTTP errors (like 422) that contain JSON responses.
-        try:
-            error_data = e.read().decode("utf-8")
-            error_response = json.loads(error_data)
-            return error_response, e.code, file_name, file_size, args.data_provider
-        except (json.JSONDecodeError, UnicodeDecodeError):
-            error_data = e.read().decode("utf-8", errors="ignore")
-            return {"error": error_data}, e.code, file_name, file_size, args.data_provider
+    with urllib.request.urlopen(request) as response:
+        data = response.read().decode("utf-8")
+        response_data = json.loads(data)
+        return response_data, response.status, file_name, file_size, args.data_provider
 
 
 def copy_file(args):
@@ -187,20 +170,10 @@ def copy_file(args):
     request = urllib.request.Request(
         change_provider_endpoint, data=json_data, headers=headers, method="POST"
     )
-    try:
-        with urllib.request.urlopen(request) as response:
-            data = response.read().decode("utf-8")
-            response_data = json.loads(data)
-            return response_data, response.status
-
-    except urllib.error.HTTPError as e:
-        try:
-            error_data = e.read().decode("utf-8")
-            error_response = json.loads(error_data)
-            return error_response, e.code
-        except (json.JSONDecodeError, UnicodeDecodeError):
-            error_data = e.read().decode("utf-8", errors="ignore")
-            return {"error": error_data}, e.code
+    with urllib.request.urlopen(request) as response:
+        data = response.read().decode("utf-8")
+        response_data = json.loads(data)
+        return response_data, response.status
 
 
 def move_file(args):
@@ -248,20 +221,10 @@ def move_file(args):
         change_provider_endpoint, data=json_data, headers=headers, method="POST"
     )
 
-    try:
-        with urllib.request.urlopen(request) as response:
-            data = response.read().decode("utf-8")
-            response_data = json.loads(data)
-            return response_data, response.status
-
-    except urllib.error.HTTPError as e:
-        try:
-            error_data = e.read().decode("utf-8")
-            error_response = json.loads(error_data)
-            return error_response, e.code
-        except (json.JSONDecodeError, UnicodeDecodeError):
-            error_data = e.read().decode("utf-8", errors="ignore")
-            return {"error": error_data}, e.code
+    with urllib.request.urlopen(request) as response:
+        data = response.read().decode("utf-8")
+        response_data = json.loads(data)
+        return response_data, response.status
 
 
 def list_files(args):
@@ -345,20 +308,7 @@ def delete_file(args):
         delete_endpoint, data=json_data, headers=headers, method="DELETE"
     )
 
-    try:
-        with urllib.request.urlopen(request) as response:
-            data = response.read().decode("utf-8")
-            try:
-                response_data = json.loads(data)
-            except json.JSONDecodeError:
-                response_data = data
-            return response_data
-
-    except urllib.error.HTTPError as e:
-        try:
-            error_data = e.read().decode("utf-8")
-            error_response = json.loads(error_data)
-            return error_response, e.code
-        except (json.JSONDecodeError, UnicodeDecodeError):
-            error_data = e.read().decode("utf-8", errors="ignore")
-            return {"error": error_data}, e.code
+    with urllib.request.urlopen(request) as response:
+        data = response.read().decode("utf-8")
+        response_data = json.loads(data)
+        return response_data