diff --git a/capture_tests/expected_captures.txt b/capture_tests/expected_captures.txt index 374289c..b127b8a 100644 --- a/capture_tests/expected_captures.txt +++ b/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) diff --git a/cbrain_cli/cli_utils.py b/cbrain_cli/cli_utils.py index cc8e7a6..c0fe5c8 100644 --- a/cbrain_cli/cli_utils.py +++ b/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. diff --git a/cbrain_cli/data/background_activities.py b/cbrain_cli/data/background_activities.py index d8973d7..110541e 100644 --- a/cbrain_cli/data/background_activities.py +++ b/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) diff --git a/cbrain_cli/data/data_providers.py b/cbrain_cli/data/data_providers.py index 56c26af..3ce7c6e 100644 --- a/cbrain_cli/data/data_providers.py +++ b/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 diff --git a/cbrain_cli/data/files.py b/cbrain_cli/data/files.py index d5a0203..c494468 100644 --- a/cbrain_cli/data/files.py +++ b/cbrain_cli/data/files.py @@ -1,11 +1,9 @@ import json import mimetypes import os -import urllib.error -import urllib.parse import urllib.request -from cbrain_cli.cli_utils import api_token, cbrain_url, pagination +from cbrain_cli.cli_utils import api_get, api_send, api_token, cbrain_url, pagination from cbrain_cli.config import auth_headers @@ -28,17 +26,7 @@ def show_file(args): if not file_id: print("Error: File ID is required") return None - - userfile_endpoint = f"{cbrain_url}/userfiles/{file_id}" - headers = auth_headers(api_token) - - request = urllib.request.Request(userfile_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - file_data = json.loads(data) - - return file_data + return api_get(f"{cbrain_url}/userfiles/{file_id}", api_token) def upload_file(args): @@ -64,121 +52,69 @@ def upload_file(args): print("Error: Group ID is required") return None - # Get group_id from args - group_id = args.group_id - - # Get file information. file_name = os.path.basename(args.file_path) file_size = os.path.getsize(args.file_path) - # Determine MIME type. mime_type, _ = mimetypes.guess_type(args.file_path) if not mime_type: mime_type = "application/octet-stream" - # Create multipart form data. boundary = "----formdata-cbrain-cli" + body_parts = [ + f"--{boundary}", + 'Content-Disposition: form-data; name="data_provider_id"', + "", + str(args.data_provider), + f"--{boundary}", + 'Content-Disposition: form-data; name="userfile[group_id]"', + "", + str(args.group_id), + f"--{boundary}", + f'Content-Disposition: form-data; name="upload_file"; filename="{file_name}"', + f"Content-Type: {mime_type}", + "", + ] - # Build the multipart body. - body_parts = [] - - # Add data_provider_id field. - body_parts.append(f"--{boundary}") - body_parts.append('Content-Disposition: form-data; name="data_provider_id"') - body_parts.append("") - body_parts.append(str(args.data_provider)) - - # Add userfile[group_id] field. - body_parts.append(f"--{boundary}") - body_parts.append('Content-Disposition: form-data; name="userfile[group_id]"') - body_parts.append("") - body_parts.append(str(group_id)) - - # Add file data. - body_parts.append(f"--{boundary}") - body_parts.append(f'Content-Disposition: form-data; name="upload_file"; filename="{file_name}"') - body_parts.append(f"Content-Type: {mime_type}") - body_parts.append("") - - # Join the text parts. body_text = "\r\n".join(body_parts) + "\r\n" - - # Read file content. with open(args.file_path, "rb") as f: file_content = f.read() - # Complete the multipart body. - body_end = f"\r\n--{boundary}--\r\n" - - body = body_text.encode("utf-8") + file_content + body_end.encode("utf-8") + body = body_text.encode("utf-8") + file_content + f"\r\n--{boundary}--\r\n".encode() - # Prepare headers. headers = auth_headers(api_token) headers["Content-Type"] = f"multipart/form-data; boundary={boundary}" headers["Content-Length"] = str(len(body)) - # Create the request. - upload_endpoint = f"{cbrain_url}/userfiles" - request = urllib.request.Request(upload_endpoint, data=body, headers=headers, method="POST") - + request = urllib.request.Request( + f"{cbrain_url}/userfiles", data=body, headers=headers, method="POST" + ) with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) + response_data = json.loads(response.read().decode("utf-8")) return response_data, response.status, file_name, file_size, args.data_provider -def copy_file(args): - """ - Copy files to a different data provider. - - Parameters - ---------- - args : argparse.Namespace - Command line arguments, including file-id (list) and dp-id - - Returns - ------- - tuple - (response_data, response_status) or None if error - """ - # Get the file IDs and destination data provider ID +def _change_provider(args, operation): file_ids = getattr(args, "file_id", None) dest_provider_id = getattr(args, "dp_id", None) or getattr( args, "data_provider_id_for_mv_cp", None ) - if not file_ids: print("Error: File ID(s) are required") return None - if not dest_provider_id: print("Error: Destination data provider ID is required") return None - - change_provider_endpoint = f"{cbrain_url}/userfiles/change_provider" - headers = auth_headers(api_token) - headers["Content-Type"] = "application/json" - payload = { "file_ids": file_ids, "data_provider_id_for_mv_cp": dest_provider_id, - "copy": "", # Empty string indicates copy operation + operation: "", } + return api_send(f"{cbrain_url}/userfiles/change_provider", api_token, payload=payload) - json_data = json.dumps(payload).encode("utf-8") - - request = urllib.request.Request( - change_provider_endpoint, data=json_data, headers=headers, method="POST" - ) - 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): +def copy_file(args): """ - Move files to a different data provider. + Copy files to a different data provider. Parameters ---------- @@ -190,41 +126,24 @@ def move_file(args): tuple (response_data, response_status) or None if error """ - # Get the file IDs and destination data provider ID - file_ids = getattr(args, "file_id", None) - dest_provider_id = getattr(args, "dp_id", None) or getattr( - args, "data_provider_id_for_mv_cp", None - ) - - if not file_ids: - print("Error: File ID(s) are required") - return None + return _change_provider(args, "copy") - if not dest_provider_id: - print("Error: Destination data provider ID is required") - return None - change_provider_endpoint = f"{cbrain_url}/userfiles/change_provider" - headers = auth_headers(api_token) - headers["Content-Type"] = "application/json" - - # Prepare the payload for move operation - payload = { - "file_ids": file_ids, - "data_provider_id_for_mv_cp": dest_provider_id, - "move": "", # "move" key indicates move operation - } - - json_data = json.dumps(payload).encode("utf-8") +def move_file(args): + """ + Move files to a different data provider. - request = urllib.request.Request( - change_provider_endpoint, data=json_data, headers=headers, method="POST" - ) + Parameters + ---------- + args : argparse.Namespace + Command line arguments, including file-id (list) and dp-id - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - return response_data, response.status + Returns + ------- + tuple + (response_data, response_status) or None if error + """ + return _change_provider(args, "move") def list_files(args): @@ -238,42 +157,25 @@ def list_files(args): Returns ------- - tuple - (files_data, page) or None if error + list or None + List of file dictionaries, or None if error """ - # Build query parameters for filtering - query_params = {} - - # Add filter parameters if provided - if hasattr(args, "group_id") and args.group_id is not None: - query_params["group_id"] = str(args.group_id) - - if hasattr(args, "dp_id") and args.dp_id is not None: - query_params["data_provider_id"] = str(args.dp_id) - - if hasattr(args, "user_id") and args.user_id is not None: - query_params["user_id"] = str(args.user_id) - - if hasattr(args, "parent_id") and args.parent_id is not None: - query_params["parent_id"] = str(args.parent_id) - - if hasattr(args, "file_type") and args.file_type is not None: - query_params["type"] = args.file_type - - query_params = pagination(args, query_params) - - userfiles_endpoint = f"{cbrain_url}/userfiles" - query_string = urllib.parse.urlencode(query_params) - userfiles_endpoint = f"{userfiles_endpoint}?{query_string}" - - headers = auth_headers(api_token) - request = urllib.request.Request(userfiles_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - files_data = json.loads(data) - - return files_data + params = {} + for attr, key in [ + ("group_id", "group_id"), + ("dp_id", "data_provider_id"), + ("user_id", "user_id"), + ("parent_id", "parent_id"), + ("file_type", "type"), + ]: + val = getattr(args, attr, None) + if val is not None: + params[key] = str(val) + + params = pagination(args, params) + if params is None: + return None + return api_get(f"{cbrain_url}/userfiles", api_token, params) def delete_file(args): @@ -287,28 +189,17 @@ def delete_file(args): Returns ------- - tuple - (response_data, response_status) or None if error + dict or None + Response data, or None if error """ - # Get the file ID from the argument. file_id = getattr(args, "file_id", None) if not file_id: print("Error: File ID is required") return None - - delete_endpoint = f"{cbrain_url}/userfiles/delete_files" - headers = auth_headers(api_token) - headers["Content-Type"] = "application/json" - - payload = {"file_ids": [str(file_id)]} - - json_data = json.dumps(payload).encode("utf-8") - - request = urllib.request.Request( - delete_endpoint, data=json_data, headers=headers, method="DELETE" + data, _ = api_send( + f"{cbrain_url}/userfiles/delete_files", + api_token, + method="DELETE", + payload={"file_ids": [str(file_id)]}, ) - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - return response_data + return data diff --git a/cbrain_cli/data/projects.py b/cbrain_cli/data/projects.py index 58ff27b..28244a5 100644 --- a/cbrain_cli/data/projects.py +++ b/cbrain_cli/data/projects.py @@ -1,9 +1,8 @@ import json import urllib.error -import urllib.request -from cbrain_cli.cli_utils import api_token, cbrain_url -from cbrain_cli.config import CREDENTIALS_FILE, auth_headers +from cbrain_cli.cli_utils import api_get, api_send, api_token, cbrain_url +from cbrain_cli.config import CREDENTIALS_FILE def switch_project(args): @@ -26,47 +25,30 @@ def switch_project(args): print("Error: Group ID is required") return None - # Handle the special case of "all" if group_id == "all": print("Project switch 'all' not yet implemented as of Aug 2025") return None - # Convert to integer for regular group IDs try: group_id = int(group_id) except ValueError: print(f"Error: Invalid group ID '{group_id}'. Must be a number or 'all'") return None - # Step 1: Call the switch API - switch_endpoint = f"{cbrain_url}/groups/switch?id={group_id}" - headers = auth_headers(api_token) + api_send(f"{cbrain_url}/groups/switch?id={group_id}", api_token) + group_data = api_get(f"{cbrain_url}/groups/{group_id}", api_token) - # Create the request - request = urllib.request.Request(switch_endpoint, data=None, headers=headers, method="POST") - - with urllib.request.urlopen(request): - group_endpoint = f"{cbrain_url}/groups/{group_id}" - group_request = urllib.request.Request( - group_endpoint, data=None, headers=headers, method="GET" - ) - - with urllib.request.urlopen(group_request) as group_response: - group_data_text = group_response.read().decode("utf-8") - group_data = json.loads(group_data_text) - - # Step 3: Update credentials file with current group_id - if CREDENTIALS_FILE.exists(): - with open(CREDENTIALS_FILE) as f: - credentials = json.load(f) + if CREDENTIALS_FILE.exists(): + with open(CREDENTIALS_FILE) as f: + credentials = json.load(f) - credentials["current_group_id"] = group_id - credentials["current_group_name"] = group_data.get("name", "Unknown") + credentials["current_group_id"] = group_id + credentials["current_group_name"] = group_data.get("name", "Unknown") - with open(CREDENTIALS_FILE, "w") as f: - json.dump(credentials, f, indent=2) + with open(CREDENTIALS_FILE, "w") as f: + json.dump(credentials, f, indent=2) - return group_data + return group_data def show_project(args): @@ -87,54 +69,33 @@ def show_project(args): project_id = getattr(args, "project_id", None) if project_id: - # Show specific project by ID - group_endpoint = f"{cbrain_url}/groups/{project_id}" - headers = auth_headers(api_token) - request = urllib.request.Request(group_endpoint, data=None, headers=headers, method="GET") - + # Show specific project by ID try: - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - group_data = json.loads(data) - return group_data + return api_get(f"{cbrain_url}/groups/{project_id}", api_token) except urllib.error.HTTPError as e: if e.code == 404: print(f"Error: Project with ID {project_id} not found") return None - else: - raise - else: - # Show current project from credentials - with open(CREDENTIALS_FILE) as f: - credentials = json.load(f) - - current_group_id = credentials.get("current_group_id") - if not current_group_id: - return None - - # Get fresh group details from server - group_endpoint = f"{cbrain_url}/groups/{current_group_id}" - headers = auth_headers(api_token) + raise - request = urllib.request.Request(group_endpoint, data=None, headers=headers, method="GET") + with open(CREDENTIALS_FILE) as f: + credentials = json.load(f) - try: - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - group_data = json.loads(data) - return group_data + current_group_id = credentials.get("current_group_id") + if not current_group_id: + return None - except urllib.error.HTTPError as e: - if e.code == 404: - print(f"Error: Current project (ID {current_group_id}) no longer exists") - # Clear the invalid group_id from credentials - credentials.pop("current_group_id", None) - credentials.pop("current_group_name", None) - with open(CREDENTIALS_FILE, "w") as f: - json.dump(credentials, f, indent=2) - return None - else: - raise + try: + return api_get(f"{cbrain_url}/groups/{current_group_id}", api_token) + except urllib.error.HTTPError as e: + if e.code == 404: + print(f"Error: Current project (ID {current_group_id}) no longer exists") + credentials.pop("current_group_id", None) + credentials.pop("current_group_name", None) + with open(CREDENTIALS_FILE, "w") as f: + json.dump(credentials, f, indent=2) + return None + raise def list_projects(args): @@ -151,16 +112,4 @@ def list_projects(args): list List of project dictionaries """ - # Prepare the API request. - groups_endpoint = f"{cbrain_url}/groups" - headers = auth_headers(api_token) - - # Create the request. - request = urllib.request.Request(groups_endpoint, data=None, headers=headers, method="GET") - - # Make the request. - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - projects_data = json.loads(data) - - return projects_data + return api_get(f"{cbrain_url}/groups", api_token) diff --git a/cbrain_cli/data/remote_resources.py b/cbrain_cli/data/remote_resources.py index 9133992..96cab52 100644 --- a/cbrain_cli/data/remote_resources.py +++ b/cbrain_cli/data/remote_resources.py @@ -1,9 +1,4 @@ -import json -import urllib.error -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_remote_resources(args): @@ -20,16 +15,7 @@ def list_remote_resources(args): list List of remote resource dictionaries """ - bourreaux_endpoint = f"{cbrain_url}/bourreaux" - headers = auth_headers(api_token) - - request = urllib.request.Request(bourreaux_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - bourreaux_data = json.loads(data) - - return bourreaux_data + return api_get(f"{cbrain_url}/bourreaux", api_token) def show_remote_resource(args): @@ -46,18 +32,8 @@ def show_remote_resource(args): dict or None Dictionary containing remote resource details if successful, None otherwise """ - # Get the remote resource ID from the remote_resource argument resource_id = getattr(args, "remote_resource", None) if not resource_id: print("Error: Remote resource ID is required") return None - - bourreau_endpoint = f"{cbrain_url}/bourreaux/{resource_id}" - headers = auth_headers(api_token) - - request = urllib.request.Request(bourreau_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - bourreau_data = json.loads(data) - return bourreau_data + return api_get(f"{cbrain_url}/bourreaux/{resource_id}", api_token) diff --git a/cbrain_cli/data/tags.py b/cbrain_cli/data/tags.py index 1105cc4..ff6ed4d 100644 --- a/cbrain_cli/data/tags.py +++ b/cbrain_cli/data/tags.py @@ -1,10 +1,18 @@ -import json -import urllib.error -import urllib.parse -import urllib.request +from cbrain_cli.cli_utils import api_get, api_send, api_token, cbrain_url, pagination -from cbrain_cli.cli_utils import api_token, cbrain_url, pagination -from cbrain_cli.config import auth_headers +tags = [ + ("name", "Tag name", "--name"), + ("user_id", "User ID", "--user-id"), + ("group_id", "Group ID", "--group-id"), +] + + +def _tag_payload(args): + for attr, label, flag in tags: + if not getattr(args, attr, None): + print(f"Error: {label} is required. Use {flag} flag") + return None + return {"tag": {"name": args.name, "user_id": args.user_id, "group_id": args.group_id}} def list_tags(args): @@ -21,21 +29,10 @@ def list_tags(args): list List of tag dictionaries """ - query_params = {} - query_params = pagination(args, query_params) - - tags_endpoint = f"{cbrain_url}/tags" - query_string = urllib.parse.urlencode(query_params) - tags_endpoint = f"{tags_endpoint}?{query_string}" - headers = auth_headers(api_token) - - request = urllib.request.Request(tags_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - tags_data = json.loads(data) - - return tags_data + params = pagination(args, {}) + if params is None: + return None + return api_get(f"{cbrain_url}/tags", api_token, params) def show_tag(args): @@ -57,16 +54,7 @@ def show_tag(args): if not tag_id: print("Error: Tag ID is required") return None - - tag_endpoint = f"{cbrain_url}/tags/{tag_id}" - headers = auth_headers(api_token) - - request = urllib.request.Request(tag_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - tag_data = json.loads(data) - return tag_data + return api_get(f"{cbrain_url}/tags/{tag_id}", api_token) def create_tag(args): @@ -84,37 +72,11 @@ def create_tag(args): (response_data, success, error_msg, response_status) """ # Get tag details from command line arguments - tag_name = getattr(args, "name", None) - user_id = getattr(args, "user_id", None) - group_id = getattr(args, "group_id", None) - - if not tag_name: - print("Error: Tag name is required. Use --name flag") - return None, False, None, None - - if not user_id: - print("Error: User ID is required. Use --user-id flag") - return None, False, None, None - - if not group_id: - print("Error: Group ID is required. Use --group-id flag") + payload = _tag_payload(args) + if payload is None: return None, False, None, None - - # Prepare the API request - tags_endpoint = f"{cbrain_url}/tags" - headers = auth_headers(api_token) - headers["Content-Type"] = "application/json" - - # Prepare the payload - payload = {"tag": {"name": tag_name, "user_id": user_id, "group_id": group_id}} - json_data = json.dumps(payload).encode("utf-8") - - request = urllib.request.Request(tags_endpoint, data=json_data, headers=headers, method="POST") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - return response_data, True, None, response.status + data, status = api_send(f"{cbrain_url}/tags", api_token, payload=payload) + return data, True, None, status def update_tag(args): @@ -136,38 +98,11 @@ def update_tag(args): if not tag_id: print("Error: Tag ID is required. Provide tag_id argument") return None, False, None, None - - tag_name = getattr(args, "name", None) - user_id = getattr(args, "user_id", None) - group_id = getattr(args, "group_id", None) - - if not tag_name: - print("Error: Tag name is required. Use --name flag") - return None, False, None, None - - if not user_id: - print("Error: User ID is required. Use --user-id flag") - return None, False, None, None - - if not group_id: - print("Error: Group ID is required. Use --group-id flag") + payload = _tag_payload(args) + if payload is None: return None, False, None, None - - # Prepare the API request - tag_endpoint = f"{cbrain_url}/tags/{tag_id}" - headers = auth_headers(api_token) - headers["Content-Type"] = "application/json" - - # Prepare the payload - payload = {"tag": {"name": tag_name, "user_id": user_id, "group_id": group_id}} - json_data = json.dumps(payload).encode("utf-8") - - request = urllib.request.Request(tag_endpoint, data=json_data, headers=headers, method="PUT") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - return response_data, True, None, response.status + data, status = api_send(f"{cbrain_url}/tags/{tag_id}", api_token, method="PUT", payload=payload) + return data, True, None, status def delete_tag(args): @@ -189,12 +124,5 @@ def delete_tag(args): if not tag_id: print("Error: Tag ID is required. Provide tag_id argument") return False, None, None - - # Prepare the API request - tag_endpoint = f"{cbrain_url}/tags/{tag_id}" - headers = auth_headers(api_token) - - request = urllib.request.Request(tag_endpoint, data=None, headers=headers, method="DELETE") - - with urllib.request.urlopen(request) as response: - return True, None, response.status + _, status = api_send(f"{cbrain_url}/tags/{tag_id}", api_token, method="DELETE") + return True, None, status diff --git a/cbrain_cli/data/tasks.py b/cbrain_cli/data/tasks.py index 58e5453..6460ac3 100644 --- a/cbrain_cli/data/tasks.py +++ b/cbrain_cli/data/tasks.py @@ -1,9 +1,4 @@ -import json -import urllib.parse -import urllib.request - -from cbrain_cli.cli_utils import api_token, cbrain_url, json_printer, pagination -from cbrain_cli.config import auth_headers +from cbrain_cli.cli_utils import api_get, api_send, api_token, cbrain_url, json_printer, pagination def list_tasks(args): @@ -17,40 +12,25 @@ def list_tasks(args): Returns ------- - int - Exit code (0 for success, 1 for failure) + list or None + List of task dictionaries, or None on error """ - # Build query parameters for filtering. - query_params = {} + params = {} - # Add filter if provided. if hasattr(args, "filter_type") and args.filter_type is not None: if args.filter_value is None: print("Error: Filter value is required when filter type is specified") - return 1 + return None if args.filter_type == "bourreau_id": - query_params["bourreau_id"] = str(args.filter_value) + params["bourreau_id"] = str(args.filter_value) elif hasattr(args, "filter_value") and args.filter_value is not None: print("Error: Filter type is required when filter value is specified") - return 1 - - query_params = pagination(args, query_params) - - tasks_endpoint = f"{cbrain_url}/tasks" - - if query_params: - query_string = urllib.parse.urlencode(query_params) - tasks_endpoint = f"{tasks_endpoint}?{query_string}" - - headers = auth_headers(api_token) - - request = urllib.request.Request(tasks_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - tasks_data = json.loads(data) + return None - return tasks_data + params = pagination(args, params) + if params is None: + return None + return api_get(f"{cbrain_url}/tasks", api_token, params) def show_task(args): @@ -64,39 +44,19 @@ def show_task(args): Returns ------- - int - Exit code (0 for success, 1 for failure) + dict or None + Task details dictionary, or None on error """ - # Get the task ID from the task argument. task_id = getattr(args, "task", None) if not task_id: print("Error: Task ID is required") - return 1 - - task_endpoint = f"{cbrain_url}/tasks/{task_id}" - headers = auth_headers(api_token) - - request = urllib.request.Request(task_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - task_data = json.loads(data) - - return task_data + return None + return api_get(f"{cbrain_url}/tasks/{task_id}", api_token) def operation_task(args): """ Operation on a task. """ - operate_task_endpoint = f"{cbrain_url}/tasks/operation" - headers = auth_headers(api_token) - - request = urllib.request.Request( - operate_task_endpoint, data=None, headers=headers, method="POST" - ) - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - parsed_data = json.loads(data) - json_printer(parsed_data) + data, _ = api_send(f"{cbrain_url}/tasks/operation", api_token) + json_printer(data) diff --git a/cbrain_cli/data/tool_configs.py b/cbrain_cli/data/tool_configs.py index 92c16fe..c4fc8bb 100644 --- a/cbrain_cli/data/tool_configs.py +++ b/cbrain_cli/data/tool_configs.py @@ -1,89 +1,41 @@ -import json -import urllib.parse -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_token, cbrain_url, pagination def list_tool_configs(args): """ Lists all tool configurations available in the system. - Sends a GET request to the tool configurations endpoint to retrieve - a list of all tool configurations. The response is then parsed and returned as a JSON object. - Returns ------- list A list of tool configurations, each represented as a dictionary containing configuration details. """ - query_params = {} - query_params = pagination(args, query_params) - - tool_configs_endpoint = f"{cbrain_url}/tool_configs" - query_string = urllib.parse.urlencode(query_params) - tool_configs_endpoint = f"{tool_configs_endpoint}?{query_string}" - headers = auth_headers(api_token) - - request = urllib.request.Request( - tool_configs_endpoint, data=None, headers=headers, method="GET" - ) - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - return response_data + params = pagination(args, {}) + if params is None: + return None + return api_get(f"{cbrain_url}/tool_configs", api_token, params) def show_tool_config(args): """ Retrieves detailed information about a specific tool configuration. - Detailed information about a specific tool configuration. - Returns ------- dict A dictionary containing the detailed information for the specified tool configuration. """ - show_tool_config_endpoint = f"{cbrain_url}/tool_configs/{args.id}" - headers = auth_headers(api_token) - - request = urllib.request.Request( - show_tool_config_endpoint, data=None, headers=headers, method="GET" - ) - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - return response_data + return api_get(f"{cbrain_url}/tool_configs/{args.id}", api_token) def tool_config_boutiques_descriptor(args): """ Retrieves the Boutiques descriptor for a specific tool configuration. - Sends a GET request to the tool configuration Boutiques descriptor endpoint - to retrieve the descriptor for a specific tool configuration. - The response is then parsed and returned as a JSON object. - Returns ------- dict A dictionary containing the Boutiques descriptor for the specified tool configuration. """ - tool_config_boutiques_descriptor_endpoint = ( - f"{cbrain_url}/tool_configs/{args.id}/boutiques_descriptor" - ) - headers = auth_headers(api_token) - - request = urllib.request.Request( - tool_config_boutiques_descriptor_endpoint, data=None, headers=headers, method="GET" - ) - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - return response_data + return api_get(f"{cbrain_url}/tool_configs/{args.id}/boutiques_descriptor", api_token) diff --git a/cbrain_cli/data/tools.py b/cbrain_cli/data/tools.py index fdd8c7a..9480007 100644 --- a/cbrain_cli/data/tools.py +++ b/cbrain_cli/data/tools.py @@ -1,8 +1,4 @@ -import json -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_token, cbrain_url, pagination def list_tools(args): @@ -23,19 +19,10 @@ def list_tools(args): """ # Get the tool ID from the -id argument if provided. tool_id = getattr(args, "id", None) - query_params = {} - query_params = pagination(args, query_params) - - tools_endpoint = f"{cbrain_url}/tools" - query_string = urllib.parse.urlencode(query_params) - tools_endpoint = f"{tools_endpoint}?{query_string}" - headers = auth_headers(api_token) - - request = urllib.request.Request(tools_endpoint, data=None, headers=headers, method="GET") - - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - tools_data = json.loads(data) + params = pagination(args, {}) + if params is None: + return None + tools_data = api_get(f"{cbrain_url}/tools", api_token, params) if not isinstance(tools_data, list): print("Error: Unexpected response format from server") @@ -48,6 +35,5 @@ def list_tools(args): print(f"Error: Tool with ID {tool_id} not found") return None return tool - else: - # Return all tools - return tools_data + + return tools_data diff --git a/cbrain_cli/formatter/background_activities_fmt.py b/cbrain_cli/formatter/background_activities_fmt.py index 847d099..1a96ce1 100644 --- a/cbrain_cli/formatter/background_activities_fmt.py +++ b/cbrain_cli/formatter/background_activities_fmt.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, display_key_value_table, output_json def print_activities_list(activities_data, args): @@ -12,39 +12,32 @@ def print_activities_list(activities_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(activities_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(activities_data) + if output_json(args, activities_data): return if not activities_data: print("No background activities found.") return - # Prepare data with formatted timestamps and items for better display - formatted_activities = [] - for activity in activities_data: - created_at = activity.get("created_at", "") - # Format created_at to show only date and time without timezone - if created_at: - created_at = created_at.split("T")[0] + " " + created_at.split("T")[1].split(".")[0] - - items = activity.get("items", []) - items_str = ",".join(map(str, items)) if items else "" - - formatted_activity = { - "id": activity.get("id", ""), - "user_id": activity.get("user_id", ""), - "remote_resource_id": activity.get("remote_resource_id", ""), - "status": activity.get("status", ""), - "created_at": created_at, - "items": items_str, - "num_successes": activity.get("num_successes", 0), - "num_failures": activity.get("num_failures", 0), + formatted_activities = [ + { + "id": a.get("id", ""), + "user_id": a.get("user_id", ""), + "remote_resource_id": a.get("remote_resource_id", ""), + "status": a.get("status", ""), + "created_at": ( + a.get("created_at", "").split("T")[0] + + " " + + a.get("created_at", "").split("T")[1].split(".")[0] + if a.get("created_at") + else "" + ), + "items": ",".join(map(str, a.get("items", []))) if a.get("items") else "", + "num_successes": a.get("num_successes", 0), + "num_failures": a.get("num_failures", 0), } - formatted_activities.append(formatted_activity) + for a in activities_data + ] dynamic_table_print( formatted_activities, @@ -73,56 +66,46 @@ def print_activity_details(activity_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(activity_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(activity_data) + if output_json(args, activity_data): return print("BACKGROUND ACTIVITY DETAILS") print("-" * 30) - - basic_info = [ - {"field": "ID", "value": str(activity_data.get("id", "N/A"))}, - {"field": "Type", "value": str(activity_data.get("type", "N/A"))}, - {"field": "User ID", "value": str(activity_data.get("user_id", "N/A"))}, - { - "field": "Remote Resource ID", - "value": str(activity_data.get("remote_resource_id", "N/A")), - }, - {"field": "Status", "value": str(activity_data.get("status", "N/A"))}, - ] - - dynamic_table_print(basic_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("ID", str(activity_data.get("id", "N/A"))), + ("Type", str(activity_data.get("type", "N/A"))), + ("User ID", str(activity_data.get("user_id", "N/A"))), + ("Remote Resource ID", str(activity_data.get("remote_resource_id", "N/A"))), + ("Status", str(activity_data.get("status", "N/A"))), + ] + ) print() print("EXECUTION INFO") print("-" * 30) - - execution_info = [ - {"field": "Handler Lock", "value": str(activity_data.get("handler_lock", "N/A"))}, - {"field": "Items", "value": str(activity_data.get("items", []))}, - {"field": "Current Item", "value": str(activity_data.get("current_item", "N/A"))}, - {"field": "Number of Successes", "value": str(activity_data.get("num_successes", "N/A"))}, - {"field": "Number of Failures", "value": str(activity_data.get("num_failures", "N/A"))}, - {"field": "Messages", "value": str(activity_data.get("messages", []))}, - {"field": "Options", "value": str(activity_data.get("options", {}))}, - ] - - dynamic_table_print(execution_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("Handler Lock", str(activity_data.get("handler_lock", "N/A"))), + ("Items", str(activity_data.get("items", []))), + ("Current Item", str(activity_data.get("current_item", "N/A"))), + ("Number of Successes", str(activity_data.get("num_successes", "N/A"))), + ("Number of Failures", str(activity_data.get("num_failures", "N/A"))), + ("Messages", str(activity_data.get("messages", []))), + ("Options", str(activity_data.get("options", {}))), + ] + ) print() print("SCHEDULING INFO") print("-" * 30) - - scheduling_info = [ - {"field": "Created At", "value": str(activity_data.get("created_at", "N/A"))}, - {"field": "Updated At", "value": str(activity_data.get("updated_at", "N/A"))}, - {"field": "Start At", "value": str(activity_data.get("start_at", "N/A"))}, - {"field": "Repeat", "value": str(activity_data.get("repeat", "N/A"))}, - {"field": "Retry Count", "value": str(activity_data.get("retry_count", "N/A"))}, - {"field": "Retry Delay", "value": str(activity_data.get("retry_delay", "N/A"))}, - ] - - dynamic_table_print(scheduling_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("Created At", str(activity_data.get("created_at", "N/A"))), + ("Updated At", str(activity_data.get("updated_at", "N/A"))), + ("Start At", str(activity_data.get("start_at", "N/A"))), + ("Repeat", str(activity_data.get("repeat", "N/A"))), + ("Retry Count", str(activity_data.get("retry_count", "N/A"))), + ("Retry Delay", str(activity_data.get("retry_delay", "N/A"))), + ] + ) diff --git a/cbrain_cli/formatter/data_providers_fmt.py b/cbrain_cli/formatter/data_providers_fmt.py index 7586812..4f80fc6 100644 --- a/cbrain_cli/formatter/data_providers_fmt.py +++ b/cbrain_cli/formatter/data_providers_fmt.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, display_key_value_table, output_json def print_provider_details(provider_data, args): @@ -12,63 +12,50 @@ def print_provider_details(provider_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(provider_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(provider_data) + if output_json(args, provider_data): return print("DATA PROVIDER DETAILS") print("-" * 30) - - # Basic information - basic_info = [ - {"field": "ID", "value": str(provider_data.get("id", "N/A"))}, - {"field": "Name", "value": str(provider_data.get("name", "N/A"))}, - {"field": "Type", "value": str(provider_data.get("type", "N/A"))}, - {"field": "Description", "value": str(provider_data.get("description", "N/A"))}, - ] - - dynamic_table_print(basic_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("ID", str(provider_data.get("id", "N/A"))), + ("Name", str(provider_data.get("name", "N/A"))), + ("Type", str(provider_data.get("type", "N/A"))), + ("Description", str(provider_data.get("description", "N/A"))), + ] + ) print() print("CONNECTION INFO") print("-" * 30) - - # Connection information - connection_info = [ - {"field": "Remote User", "value": str(provider_data.get("remote_user", "N/A"))}, - {"field": "Remote Host", "value": str(provider_data.get("remote_host", "N/A"))}, - {"field": "Remote Directory", "value": str(provider_data.get("remote_dir", "N/A"))}, - {"field": "Remote Port", "value": str(provider_data.get("remote_port", "N/A"))}, - ] - - dynamic_table_print(connection_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("Remote User", str(provider_data.get("remote_user", "N/A"))), + ("Remote Host", str(provider_data.get("remote_host", "N/A"))), + ("Remote Directory", str(provider_data.get("remote_dir", "N/A"))), + ("Remote Port", str(provider_data.get("remote_port", "N/A"))), + ] + ) print() print("OWNERSHIP & STATUS") print("-" * 30) - - # Ownership and status information - status_info = [ - {"field": "User ID", "value": str(provider_data.get("user_id", "N/A"))}, - {"field": "Group ID", "value": str(provider_data.get("group_id", "N/A"))}, - {"field": "Online", "value": str(provider_data.get("online", "N/A"))}, - {"field": "Read Only", "value": str(provider_data.get("read_only", "N/A"))}, - {"field": "Is Browsable", "value": str(provider_data.get("is_browsable", "N/A"))}, - {"field": "Is Fast Syncing", "value": str(provider_data.get("is_fast_syncing", "N/A"))}, - { - "field": "Allow File Owner Change", - "value": str(provider_data.get("allow_file_owner_change", "N/A")), - }, - { - "field": "Content Storage Shared Between Users", - "value": str(provider_data.get("content_storage_shared_between_users", "N/A")), - }, - ] - - dynamic_table_print(status_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("User ID", str(provider_data.get("user_id", "N/A"))), + ("Group ID", str(provider_data.get("group_id", "N/A"))), + ("Online", str(provider_data.get("online", "N/A"))), + ("Read Only", str(provider_data.get("read_only", "N/A"))), + ("Is Browsable", str(provider_data.get("is_browsable", "N/A"))), + ("Is Fast Syncing", str(provider_data.get("is_fast_syncing", "N/A"))), + ("Allow File Owner Change", str(provider_data.get("allow_file_owner_change", "N/A"))), + ( + "Content Storage Shared Between Users", + str(provider_data.get("content_storage_shared_between_users", "N/A")), + ), + ] + ) def print_providers_list(providers_data, args): @@ -82,27 +69,23 @@ def print_providers_list(providers_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(providers_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(providers_data) + if output_json(args, providers_data): return if not providers_data: print("No data providers found.") return - formatted_providers = [] - for provider in providers_data: - formatted_provider = { - "id": provider.get("id", ""), - "name": provider.get("name", ""), - "type": provider.get("type", ""), - "remote_host": provider.get("remote_host", ""), - "online": "Yes" if provider.get("online", False) else "No", + formatted_providers = [ + { + "id": p.get("id", ""), + "name": p.get("name", ""), + "type": p.get("type", ""), + "remote_host": p.get("remote_host", ""), + "online": "Yes" if p.get("online", False) else "No", } - formatted_providers.append(formatted_provider) + for p in providers_data + ] dynamic_table_print( formatted_providers, diff --git a/cbrain_cli/formatter/files_fmt.py b/cbrain_cli/formatter/files_fmt.py index 8946b56..9b99934 100644 --- a/cbrain_cli/formatter/files_fmt.py +++ b/cbrain_cli/formatter/files_fmt.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, output_json def print_file_details(file_data, args): @@ -12,12 +12,9 @@ def print_file_details(file_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(file_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(file_data) + if output_json(args, file_data): return + print( f"id: {file_data.get('id', 'N/A')}\n" f"type: {file_data.get('type', 'N/A')}\n" @@ -50,11 +47,7 @@ def print_files_list(files_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(files_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(files_data) + if output_json(args, files_data): return # Use the reusable dynamic table formatter @@ -125,11 +118,7 @@ def print_delete_result(response_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(response_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(response_data) + if output_json(args, response_data): return # Show user-friendly message for normal output diff --git a/cbrain_cli/formatter/projects_fmt.py b/cbrain_cli/formatter/projects_fmt.py index ef778b2..b721669 100644 --- a/cbrain_cli/formatter/projects_fmt.py +++ b/cbrain_cli/formatter/projects_fmt.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, display_key_value_table, output_json def print_projects_list(projects_data, args): @@ -13,22 +13,13 @@ def print_projects_list(projects_data, args): Command line arguments, including the --json flag """ formatted_data = [ - { - "id": project.get("id"), - "type": project.get("type"), - "name": project.get("name"), - } - for project in projects_data + {"id": p.get("id"), "type": p.get("type"), "name": p.get("name")} for p in projects_data ] - if getattr(args, "json", False): - json_printer(formatted_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(formatted_data) + if output_json(args, formatted_data): return - dynamic_table_print(projects_data, ["id", "type", "name"], ["ID", "Type", "Project Name"]) + dynamic_table_print(formatted_data, ["id", "type", "name"], ["ID", "Type", "Project Name"]) def print_current_project(project_data): @@ -56,36 +47,27 @@ def print_project_details(project_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(project_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(project_data) + if output_json(args, project_data): return print("PROJECT DETAILS") print("-" * 30) + display_key_value_table( + [ + ("ID", str(project_data.get("id", "N/A"))), + ("Name", str(project_data.get("name", "N/A"))), + ("Type", str(project_data.get("type", "N/A"))), + ("Site ID", str(project_data.get("site_id", "N/A"))), + ("Invisible", str(project_data.get("invisible", "N/A"))), + ] + ) - # Basic project information - basic_info = [ - {"field": "ID", "value": str(project_data.get("id", "N/A"))}, - {"field": "Name", "value": str(project_data.get("name", "N/A"))}, - {"field": "Type", "value": str(project_data.get("type", "N/A"))}, - {"field": "Site ID", "value": str(project_data.get("site_id", "N/A"))}, - {"field": "Invisible", "value": str(project_data.get("invisible", "N/A"))}, - ] - - dynamic_table_print(basic_info, ["field", "value"], ["Field", "Value"]) - - # Display description if available if project_data.get("description"): print() print("DESCRIPTION") print("-" * 30) - description = project_data.get("description").strip() - # Handle multi-line descriptions - for line in description.split("\n"): - print(f"{line}") + for line in project_data.get("description").strip().split("\n"): + print(line) def print_no_project(): diff --git a/cbrain_cli/formatter/remote_resources_fmt.py b/cbrain_cli/formatter/remote_resources_fmt.py index eb2e184..61b3a69 100644 --- a/cbrain_cli/formatter/remote_resources_fmt.py +++ b/cbrain_cli/formatter/remote_resources_fmt.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, display_key_value_table, output_json def print_resources_list(resources_data, args): @@ -12,11 +12,7 @@ def print_resources_list(resources_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(resources_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(resources_data) + if output_json(args, resources_data): return if not resources_data: @@ -26,18 +22,17 @@ def print_resources_list(resources_data, args): print("REMOTE RESOURCES (EXECUTION SERVERS)") print("-" * 80) - # Prepare data with formatted boolean values for better display - formatted_resources = [] - for resource in resources_data: - formatted_resource = { - "id": resource.get("id", ""), - "name": resource.get("name", ""), - "user_id": resource.get("user_id", ""), - "group_id": resource.get("group_id", ""), - "online": "Yes" if resource.get("online", False) else "No", - "read_only": "Yes" if resource.get("read_only", False) else "No", + formatted_resources = [ + { + "id": r.get("id", ""), + "name": r.get("name", ""), + "user_id": r.get("user_id", ""), + "group_id": r.get("group_id", ""), + "online": "Yes" if r.get("online", False) else "No", + "read_only": "Yes" if r.get("read_only", False) else "No", } - formatted_resources.append(formatted_resource) + for r in resources_data + ] dynamic_table_print( formatted_resources, @@ -60,36 +55,28 @@ def print_resource_details(resource_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(resource_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(resource_data) + if output_json(args, resource_data): return print("REMOTE RESOURCE DETAILS") print("-" * 30) - - # Prepare basic details as key-value pairs for table display - basic_details = [ - {"field": "ID", "value": str(resource_data.get("id", "N/A"))}, - {"field": "Name", "value": str(resource_data.get("name", "N/A"))}, - {"field": "Type", "value": str(resource_data.get("type", "N/A"))}, - ] - - dynamic_table_print(basic_details, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("ID", str(resource_data.get("id", "N/A"))), + ("Name", str(resource_data.get("name", "N/A"))), + ("Type", str(resource_data.get("type", "N/A"))), + ] + ) print() print("OWNERSHIP & ACCESS") print("-" * 30) - - # Prepare ownership details as key-value pairs for table display - ownership_details = [ - {"field": "User ID", "value": str(resource_data.get("user_id", "N/A"))}, - {"field": "Group ID", "value": str(resource_data.get("group_id", "N/A"))}, - {"field": "Online", "value": str(resource_data.get("online", "N/A"))}, - {"field": "Read Only", "value": str(resource_data.get("read_only", "N/A"))}, - ] - - dynamic_table_print(ownership_details, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("User ID", str(resource_data.get("user_id", "N/A"))), + ("Group ID", str(resource_data.get("group_id", "N/A"))), + ("Online", str(resource_data.get("online", "N/A"))), + ("Read Only", str(resource_data.get("read_only", "N/A"))), + ] + ) print() diff --git a/cbrain_cli/formatter/tags_fmt.py b/cbrain_cli/formatter/tags_fmt.py index 29d9d4b..09c7aa0 100644 --- a/cbrain_cli/formatter/tags_fmt.py +++ b/cbrain_cli/formatter/tags_fmt.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, display_key_value_table, output_json def print_tags_list(tags_data, args): @@ -12,11 +12,7 @@ def print_tags_list(tags_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(tags_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(tags_data) + if output_json(args, tags_data): return if not tags_data: @@ -25,12 +21,9 @@ def print_tags_list(tags_data, args): print("TAGS") print("-" * 40) - - # Use the reusable dynamic table formatter dynamic_table_print( tags_data, ["id", "name", "user_id", "group_id"], ["ID", "Name", "User", "Group"] ) - print("-" * 40) print(f"Total: {len(tags_data)} tag(s)") @@ -46,25 +39,19 @@ def print_tag_details(tag_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(tag_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(tag_data) + if output_json(args, tag_data): return print("TAG DETAILS") print("-" * 30) - - # Prepare tag details as key-value pairs for table display - tag_details = [ - {"field": "ID", "value": str(tag_data.get("id", "N/A"))}, - {"field": "Name", "value": str(tag_data.get("name", "N/A"))}, - {"field": "User ID", "value": str(tag_data.get("user_id", "N/A"))}, - {"field": "Group ID", "value": str(tag_data.get("group_id", "N/A"))}, - ] - - dynamic_table_print(tag_details, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("ID", str(tag_data.get("id", "N/A"))), + ("Name", str(tag_data.get("name", "N/A"))), + ("User ID", str(tag_data.get("user_id", "N/A"))), + ("Group ID", str(tag_data.get("group_id", "N/A"))), + ] + ) def print_tag_operation_result( diff --git a/cbrain_cli/formatter/tasks_fmt.py b/cbrain_cli/formatter/tasks_fmt.py index 566db98..732d925 100644 --- a/cbrain_cli/formatter/tasks_fmt.py +++ b/cbrain_cli/formatter/tasks_fmt.py @@ -1,6 +1,6 @@ import json -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, display_key_value_table, output_json def print_task_data(tasks_data, args): @@ -14,21 +14,15 @@ def print_task_data(tasks_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(tasks_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(tasks_data) + if output_json(args, tasks_data): return if not tasks_data: print("No tasks found.") return - # Prepare data with cleaned task types for better display - formatted_tasks = [] - for task in tasks_data: - formatted_task = { + formatted_tasks = [ + { "id": task.get("id", ""), "type": task.get("type", "").replace("BoutiquesTask::", ""), "status": task.get("status", ""), @@ -36,9 +30,9 @@ def print_task_data(tasks_data, args): "user_id": task.get("user_id", ""), "group_id": task.get("group_id", ""), } - formatted_tasks.append(formatted_task) + for task in tasks_data + ] - # Use the reusable dynamic table formatter dynamic_table_print( formatted_tasks, ["id", "type", "status", "bourreau_id", "user_id", "group_id"], @@ -48,8 +42,6 @@ def print_task_data(tasks_data, args): print("-" * 85) print(f"Total: {len(tasks_data)} task(s)") - return - def print_task_details(task_data, args): """ @@ -62,81 +54,63 @@ def print_task_details(task_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(task_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(task_data) + if output_json(args, task_data): return - # Basic task info - basic_info = [ - {"field": "ID", "value": str(task_data.get("id", "N/A"))}, - {"field": "Type", "value": str(task_data.get("type", "N/A"))}, - {"field": "Status", "value": str(task_data.get("status", "N/A"))}, - ] - - dynamic_table_print(basic_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("ID", str(task_data.get("id", "N/A"))), + ("Type", str(task_data.get("type", "N/A"))), + ("Status", str(task_data.get("status", "N/A"))), + ] + ) print() print("OWNERSHIP & ASSIGNMENT") print("-" * 30) - ownership_info = [ - {"field": "User ID", "value": str(task_data.get("user_id", "N/A"))}, - {"field": "Group ID", "value": str(task_data.get("group_id", "N/A"))}, - {"field": "Bourreau ID", "value": str(task_data.get("bourreau_id", "N/A"))}, - {"field": "Tool Config ID", "value": str(task_data.get("tool_config_id", "N/A"))}, - {"field": "Batch ID", "value": str(task_data.get("batch_id", "N/A"))}, - ] - - dynamic_table_print(ownership_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("User ID", str(task_data.get("user_id", "N/A"))), + ("Group ID", str(task_data.get("group_id", "N/A"))), + ("Bourreau ID", str(task_data.get("bourreau_id", "N/A"))), + ("Tool Config ID", str(task_data.get("tool_config_id", "N/A"))), + ("Batch ID", str(task_data.get("batch_id", "N/A"))), + ] + ) print() print("EXECUTION INFO") print("-" * 30) - execution_info = [ - {"field": "Run Number", "value": str(task_data.get("run_number", "N/A"))}, - { - "field": "Results Data Provider ID", - "value": str(task_data.get("results_data_provider_id", "N/A")), - }, - { - "field": "Cluster Workdir Size", - "value": str(task_data.get("cluster_workdir_size", "N/A")), - }, - {"field": "Workdir Archived", "value": str(task_data.get("workdir_archived", "N/A"))}, - { - "field": "Workdir Archive File ID", - "value": str(task_data.get("workdir_archive_userfile_id", "N/A")), - }, - ] - - dynamic_table_print(execution_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("Run Number", str(task_data.get("run_number", "N/A"))), + ("Results Data Provider ID", str(task_data.get("results_data_provider_id", "N/A"))), + ("Cluster Workdir Size", str(task_data.get("cluster_workdir_size", "N/A"))), + ("Workdir Archived", str(task_data.get("workdir_archived", "N/A"))), + ("Workdir Archive File ID", str(task_data.get("workdir_archive_userfile_id", "N/A"))), + ] + ) print() print("TIMESTAMPS") print("-" * 30) - timestamp_info = [ - {"field": "Created At", "value": str(task_data.get("created_at", "N/A"))}, - {"field": "Updated At", "value": str(task_data.get("updated_at", "N/A"))}, - ] - - dynamic_table_print(timestamp_info, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("Created At", str(task_data.get("created_at", "N/A"))), + ("Updated At", str(task_data.get("updated_at", "N/A"))), + ] + ) - # Optional fields. if task_data.get("description"): print() print("DESCRIPTION") print("-" * 30) - description = task_data.get("description").strip() - # Handle multi-line descriptions. - for line in description.split("\n"): - print(f"{line}") + for line in task_data.get("description").strip().split("\n"): + print(line) # Display params if they exist. if task_data.get("params"): print() print("PARAMETERS") print("-" * 30) - params_json = json.dumps(task_data.get("params"), indent=2) - print(params_json) + print(json.dumps(task_data.get("params"), indent=2)) diff --git a/cbrain_cli/formatter/tool_configs_fmt.py b/cbrain_cli/formatter/tool_configs_fmt.py index bd9215d..9c5d98a 100644 --- a/cbrain_cli/formatter/tool_configs_fmt.py +++ b/cbrain_cli/formatter/tool_configs_fmt.py @@ -1,25 +1,19 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, json_printer, output_json def print_tool_configs_list(tool_configs, args): """ Pretty print a list of tool configurations. """ - if getattr(args, "json", False): - json_printer(tool_configs) - return - elif getattr(args, "jsonl", False): - jsonl_printer(tool_configs) + if output_json(args, tool_configs): return if not tool_configs: print("No tool configurations found.") return - - # Prepare data for better display - formatted_configs = [] - for config in tool_configs: - formatted_config = { + # Prepare data for better display. + formatted_configs = [ + { "id": config.get("id", ""), "version_name": config.get("version_name", ""), "tool_id": config.get("tool_id", ""), @@ -28,7 +22,8 @@ def print_tool_configs_list(tool_configs, args): "ncpus": config.get("ncpus", "1"), "description": config.get("description", ""), } - formatted_configs.append(formatted_config) + for config in tool_configs + ] dynamic_table_print( formatted_configs, @@ -56,11 +51,7 @@ def print_tool_config_details(tool_config, args): """ Pretty print the details of a tool configuration. """ - if getattr(args, "json", False): - json_printer(tool_config) - return - elif getattr(args, "jsonl", False): - jsonl_printer(tool_config) + if output_json(args, tool_config): return if not tool_config: diff --git a/cbrain_cli/formatter/tools_fmt.py b/cbrain_cli/formatter/tools_fmt.py index 946dc37..3139391 100644 --- a/cbrain_cli/formatter/tools_fmt.py +++ b/cbrain_cli/formatter/tools_fmt.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import dynamic_table_print, json_printer, jsonl_printer +from cbrain_cli.cli_utils import dynamic_table_print, display_key_value_table, output_json def print_tool_details(tool_data, args): @@ -12,28 +12,22 @@ def print_tool_details(tool_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(tool_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(tool_data) + if output_json(args, tool_data): return print("TOOL DETAILS") print("-" * 30) - - # Prepare tool details as key-value pairs for table display - tool_details = [ - {"field": "ID", "value": str(tool_data.get("id", "N/A"))}, - {"field": "Name", "value": str(tool_data.get("name", "N/A"))}, - {"field": "User ID", "value": str(tool_data.get("user_id", "N/A"))}, - {"field": "Group ID", "value": str(tool_data.get("group_id", "N/A"))}, - {"field": "Category", "value": str(tool_data.get("category", "N/A"))}, - {"field": "Description", "value": str(tool_data.get("description", "N/A"))}, - {"field": "URL", "value": str(tool_data.get("url", "N/A"))}, - ] - - dynamic_table_print(tool_details, ["field", "value"], ["Field", "Value"]) + display_key_value_table( + [ + ("ID", str(tool_data.get("id", "N/A"))), + ("Name", str(tool_data.get("name", "N/A"))), + ("User ID", str(tool_data.get("user_id", "N/A"))), + ("Group ID", str(tool_data.get("group_id", "N/A"))), + ("Category", str(tool_data.get("category", "N/A"))), + ("Description", str(tool_data.get("description", "N/A"))), + ("URL", str(tool_data.get("url", "N/A"))), + ] + ) def print_tools_list(tools_data, args): @@ -47,11 +41,7 @@ def print_tools_list(tools_data, args): args : argparse.Namespace Command line arguments, including the --json flag """ - if getattr(args, "json", False): - json_printer(tools_data) - return - elif getattr(args, "jsonl", False): - jsonl_printer(tools_data) + if output_json(args, tools_data): return if not tools_data: diff --git a/cbrain_cli/handlers.py b/cbrain_cli/handlers.py index 83c0ccf..8f99ec8 100644 --- a/cbrain_cli/handlers.py +++ b/cbrain_cli/handlers.py @@ -6,64 +6,28 @@ """ from cbrain_cli.cli_utils import json_printer -from cbrain_cli.data.background_activities import ( - list_background_activities, - show_background_activity, +from cbrain_cli.data import ( + background_activities, + data_providers, + files, + projects, + remote_resources, + tags, + tasks, + tool_configs, + tools, ) -from cbrain_cli.data.data_providers import ( - delete_unregistered_files, - is_alive, - list_data_providers, - show_data_provider, +from cbrain_cli.formatter import ( + background_activities_fmt, + data_providers_fmt, + files_fmt, + projects_fmt, + remote_resources_fmt, + tags_fmt, + tasks_fmt, + tool_configs_fmt, + tools_fmt, ) -from cbrain_cli.data.files import ( - copy_file, - delete_file, - list_files, - move_file, - show_file, - upload_file, -) -from cbrain_cli.data.projects import list_projects, show_project, switch_project -from cbrain_cli.data.remote_resources import list_remote_resources, show_remote_resource -from cbrain_cli.data.tags import create_tag, delete_tag, list_tags, show_tag, update_tag -from cbrain_cli.data.tasks import list_tasks, show_task -from cbrain_cli.data.tool_configs import ( - list_tool_configs, - show_tool_config, - tool_config_boutiques_descriptor, -) -from cbrain_cli.data.tools import list_tools -from cbrain_cli.formatter.background_activities_fmt import ( - print_activities_list, - print_activity_details, -) -from cbrain_cli.formatter.data_providers_fmt import print_provider_details, print_providers_list -from cbrain_cli.formatter.files_fmt import ( - print_delete_result, - print_file_details, - print_files_list, - print_move_copy_result, - print_upload_result, -) -from cbrain_cli.formatter.projects_fmt import ( - print_current_project, - print_no_project, - print_projects_list, -) -from cbrain_cli.formatter.remote_resources_fmt import print_resource_details, print_resources_list -from cbrain_cli.formatter.tags_fmt import ( - print_tag_details, - print_tag_operation_result, - print_tags_list, -) -from cbrain_cli.formatter.tasks_fmt import print_task_data, print_task_details -from cbrain_cli.formatter.tool_configs_fmt import ( - print_boutiques_descriptor, - print_tool_config_details, - print_tool_configs_list, -) -from cbrain_cli.formatter.tools_fmt import print_tool_details, print_tools_list # File command handlers @@ -71,106 +35,104 @@ def handle_file_list(args): """ Retrieve and display a paginated list of files from CBRAIN with optional filtering. """ - result = list_files(args) + result = files.list_files(args) if result: - print_files_list(result, args) + files_fmt.print_files_list(result, args) def handle_file_show(args): """ Retrieve and display detailed information about a specific file by its ID. """ - result = show_file(args) + result = files.show_file(args) if result: - print_file_details(result, args) + files_fmt.print_file_details(result, args) def handle_file_upload(args): """Upload a local file to CBRAIN and display the upload result with file details.""" - result = upload_file(args) + result = files.upload_file(args) if result: - print_upload_result(*result) + files_fmt.print_upload_result(*result) def handle_file_copy(args): """Copy one or more files to a different data provider and display the operation results.""" - result = copy_file(args) + result = files.copy_file(args) if result: - print_move_copy_result(*result, operation="copy") + files_fmt.print_move_copy_result(*result, operation="copy") def handle_file_move(args): """Move one or more files to a different data provider and display the operation results.""" - result = move_file(args) + result = files.move_file(args) if result: - print_move_copy_result(*result, operation="move") + files_fmt.print_move_copy_result(*result, operation="move") def handle_file_delete(args): """Delete a specific file from CBRAIN and display the deletion status.""" - result = delete_file(args) + result = files.delete_file(args) if result: - print_delete_result(result, args) + files_fmt.print_delete_result(result, args) # Data provider command handlers def handle_dataprovider_list(args): """Retrieve and display a paginated list of available data providers in CBRAIN.""" - result = list_data_providers(args) - print_providers_list(result, args) + result = data_providers.list_data_providers(args) + data_providers_fmt.print_providers_list(result, args) def handle_dataprovider_show(args): """Retrieve and display detailed information about a specific data provider.""" - result = show_data_provider(args) - print_provider_details(result, args) + result = data_providers.show_data_provider(args) + data_providers_fmt.print_provider_details(result, args) def handle_dataprovider_is_alive(args): """Check and display the connectivity status of a specific data provider.""" - result = is_alive(args) + result = data_providers.is_alive(args) json_printer(result) def handle_dataprovider_delete_unregistered(args): """Remove unregistered files from a data provider and display the cleanup results.""" - result = delete_unregistered_files(args) + result = data_providers.delete_unregistered_files(args) json_printer(result) # Project command handlers def handle_project_list(args): """Retrieve and display a list of all available projects (groups) in CBRAIN.""" - result = list_projects(args) - print_projects_list(result, args) + result = projects.list_projects(args) + projects_fmt.print_projects_list(result, args) def handle_project_switch(args): """Switch the current working context to a different project and confirm the change.""" - result = switch_project(args) + result = projects.switch_project(args) if result: - print_current_project(result) + projects_fmt.print_current_project(result) def handle_project_show(args): """Display information about the currently active project or a specific project by ID.""" - result = show_project(args) + result = projects.show_project(args) if result: # Check if a specific project ID was requested project_id = getattr(args, "project_id", None) if project_id: # Show detailed project information for specific project - from cbrain_cli.formatter.projects_fmt import print_project_details - - print_project_details(result, args) + projects_fmt.print_project_details(result, args) else: # Show current project information - print_current_project(result) + projects_fmt.print_current_project(result) else: # Only show "no project" message if no specific ID was requested project_id = getattr(args, "project_id", None) if not project_id: - print_no_project() + projects_fmt.print_no_project() def handle_project_unswitch(args): @@ -181,67 +143,67 @@ def handle_project_unswitch(args): # Tool command handlers def handle_tool_show(args): """Retrieve and display detailed information about a specific computational tool.""" - result = list_tools(args) + result = tools.list_tools(args) if result: - print_tool_details(result, args) + tools_fmt.print_tool_details(result, args) def handle_tool_list(args): """Retrieve and display a paginated list of available computational tools in CBRAIN.""" - result = list_tools(args) + result = tools.list_tools(args) if result: - print_tools_list(result, args) + tools_fmt.print_tools_list(result, args) # Tool config command handlers def handle_tool_config_list(args): """Retrieve and display a paginated list of tool configurations available in CBRAIN.""" - result = list_tool_configs(args) - print_tool_configs_list(result, args) + result = tool_configs.list_tool_configs(args) + tool_configs_fmt.print_tool_configs_list(result, args) def handle_tool_config_show(args): """Retrieve and display detailed configuration settings for a specific tool.""" - result = show_tool_config(args) + result = tool_configs.show_tool_config(args) if result: - print_tool_config_details(result, args) + tool_configs_fmt.print_tool_config_details(result, args) def handle_tool_config_boutiques_descriptor(args): """Retrieve and display the Boutiques descriptor JSON for a specific tool configuration.""" - result = tool_config_boutiques_descriptor(args) + result = tool_configs.tool_config_boutiques_descriptor(args) if result: - print_boutiques_descriptor(result, args) + tool_configs_fmt.print_boutiques_descriptor(result, args) # Tag command handlers def handle_tag_list(args): """Retrieve and display a paginated list of tags available in CBRAIN.""" - result = list_tags(args) - print_tags_list(result, args) + result = tags.list_tags(args) + tags_fmt.print_tags_list(result, args) def handle_tag_show(args): """Retrieve and display detailed information about a specific tag by its ID.""" - result = show_tag(args) + result = tags.show_tag(args) if result: - print_tag_details(result, args) + tags_fmt.print_tag_details(result, args) def handle_tag_create(args): """Create a new tag with specified name, user, and group, then display the creation result.""" - result = create_tag(args) + result = tags.create_tag(args) if result: - print_tag_operation_result( + tags_fmt.print_tag_operation_result( "create", success=result[1], error_msg=result[2], response_status=result[3] ) def handle_tag_update(args): """Update an existing tag's properties and display the modification result.""" - result = update_tag(args) + result = tags.update_tag(args) if result: - print_tag_operation_result( + tags_fmt.print_tag_operation_result( "update", tag_id=args.tag_id, success=result[1], @@ -252,9 +214,9 @@ def handle_tag_update(args): def handle_tag_delete(args): """Delete a specific tag from CBRAIN and display the deletion result.""" - result = delete_tag(args) + result = tags.delete_tag(args) if result: - print_tag_operation_result( + tags_fmt.print_tag_operation_result( "delete", tag_id=args.tag_id, success=result[0], @@ -266,41 +228,41 @@ def handle_tag_delete(args): # Background activity command handlers def handle_background_list(args): """Retrieve and display a list of background activities currently running in CBRAIN.""" - result = list_background_activities(args) + result = background_activities.list_background_activities(args) if result: - print_activities_list(result, args) + background_activities_fmt.print_activities_list(result, args) def handle_background_show(args): """Retrieve and display detailed information about a specific background activity.""" - result = show_background_activity(args) + result = background_activities.show_background_activity(args) if result: - print_activity_details(result, args) + background_activities_fmt.print_activity_details(result, args) # Task command handlers def handle_task_list(args): """Retrieve and display a paginated list of computational tasks with optional filtering.""" - result = list_tasks(args) - print_task_data(result, args) + result = tasks.list_tasks(args) + tasks_fmt.print_task_data(result, args) def handle_task_show(args): """Retrieve and display detailed information about a specific computational task.""" - result = show_task(args) + result = tasks.show_task(args) if result: - print_task_details(result, args) + tasks_fmt.print_task_details(result, args) # Remote resource command handlers def handle_remote_resource_list(args): """Retrieve and display a list of remote computational resources available in CBRAIN.""" - result = list_remote_resources(args) - print_resources_list(result, args) + result = remote_resources.list_remote_resources(args) + remote_resources_fmt.print_resources_list(result, args) def handle_remote_resource_show(args): """Retrieve and display detailed information about a specific remote computational resource.""" - result = show_remote_resource(args) + result = remote_resources.show_remote_resource(args) if result: - print_resource_details(result, args) + remote_resources_fmt.print_resource_details(result, args) diff --git a/cbrain_cli/sessions.py b/cbrain_cli/sessions.py index 4559f52..a02e6ac 100644 --- a/cbrain_cli/sessions.py +++ b/cbrain_cli/sessions.py @@ -2,16 +2,9 @@ import getpass import json import urllib.error -import urllib.parse -import urllib.request -from cbrain_cli.cli_utils import api_token, cbrain_url -from cbrain_cli.config import ( - CREDENTIALS_FILE, - DEFAULT_BASE_URL, - DEFAULT_HEADERS, - auth_headers, -) +from cbrain_cli.cli_utils import api_post_form, api_send, api_token, cbrain_url +from cbrain_cli.config import CREDENTIALS_FILE, DEFAULT_BASE_URL # MARK: Create Session. @@ -44,47 +37,27 @@ def create_session(args): print("Password is required") return 1 - # Prepare the login request. - login_endpoint = f"{cbrain_url}/session" + response_data = api_post_form(f"{cbrain_url}/session", {"login": username, "password": password}) - # Prepare form data. - form_data = {"login": username, "password": password} + cbrain_api_token = response_data.get("cbrain_api_token") + cbrain_user_id = response_data.get("user_id") - # Encode the form data. - encoded_data = urllib.parse.urlencode(form_data).encode("utf-8") - - # Create the request. - request = urllib.request.Request( - login_endpoint, data=encoded_data, headers=DEFAULT_HEADERS, method="POST" - ) - - # Make the request. - with urllib.request.urlopen(request) as response: - data = response.read().decode("utf-8") - response_data = json.loads(data) - - # Extract the API token from response. - cbrain_api_token = response_data.get("cbrain_api_token") - cbrain_user_id = response_data.get("user_id") - - if not cbrain_api_token: - print("Login failed: No API token received") - return 1 + if not cbrain_api_token: + print("Login failed: No API token received") + return 1 - # Prepare credentials data. - credentials = { - "cbrain_url": cbrain_url, - "api_token": cbrain_api_token, - "user_id": cbrain_user_id, - "timestamp": datetime.datetime.now().isoformat(), - } + credentials = { + "cbrain_url": cbrain_url, + "api_token": cbrain_api_token, + "user_id": cbrain_user_id, + "timestamp": datetime.datetime.now().isoformat(), + } - # Save credentials to file. - with open(CREDENTIALS_FILE, "w") as f: - json.dump(credentials, f, indent=2) + with open(CREDENTIALS_FILE, "w") as f: + json.dump(credentials, f, indent=2) - print(f"Connection successful, API token saved in {CREDENTIALS_FILE}") - return 0 + print(f"Connection successful, API token saved in {CREDENTIALS_FILE}") + return 0 # MARK: Logout @@ -103,27 +76,12 @@ def logout_session(args): CREDENTIALS_FILE.unlink() return 0 - # Prepare logout request. - logout_endpoint = f"{cbrain_url}/session" - - # Create headers with authorization. - headers = auth_headers(api_token) - - # Create the DELETE request. - request = urllib.request.Request( - logout_endpoint, - data=None, # No payload for DELETE - headers=headers, - method="DELETE", - ) - - # Make the request to logout from server. try: - with urllib.request.urlopen(request) as response: - if response.status == 200: - print("Successfully logged out from CBRAIN server.") - else: - print("Logout failed") + _, status = api_send(f"{cbrain_url}/session", api_token, method="DELETE") + if status == 200: + print("Successfully logged out from CBRAIN server.") + else: + print("Logout failed") except urllib.error.HTTPError as e: if e.code == 401: print("Session already expired on server.")