Created
March 1, 2024 02:50
-
-
Save Nadiar/237aa80908e71b3d93404a63a400c8ef to your computer and use it in GitHub Desktop.
Quick and easy way to browse and interact with an API, in this case the Kapowarr API. Stores successful prior attempts in a log for reference or copy pasting and running with CLI arguments
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/python3 | |
| # -*- coding: utf-8 -*- | |
| # Author github.com/Nadiar | |
| import argparse | |
| import json | |
| import requests | |
| import os | |
| from os.path import isfile, join | |
| from datetime import datetime | |
| # Constants | |
| KAPOWARR_BASE_URL = 'https://kapowarr.local/' # URI | |
| KAPOWARR_API_TOKEN = '32bitstring' # API Token | |
| LOG_FILE = "query_log.txt" | |
| def save_to_log(endpoint, method, data=None): | |
| log_entry = { | |
| "timestamp": str(datetime.now()), | |
| "endpoint": endpoint, | |
| "method": method, | |
| "data": data, | |
| } | |
| with open(LOG_FILE, "a") as log: | |
| log.write(json.dumps(log_entry) + "\n") | |
| print("Saved to log:", log_entry) # Add this line for debugging | |
| def load_prior_tasks(): | |
| prior_tasks = [] | |
| if os.path.exists(LOG_FILE): | |
| with open(LOG_FILE, "r") as log: | |
| try: | |
| for line in log: | |
| log_entry = json.loads(line) | |
| if all(key in log_entry and log_entry[key] for key in ['timestamp', 'endpoint', 'method']): | |
| prior_tasks.append(log_entry) | |
| except json.JSONDecodeError as e: | |
| print(f"Error decoding JSON in log file: {e}") | |
| return [] | |
| else: | |
| print("Log file not found. Initializing a new log.") | |
| return sorted(prior_tasks, key=lambda x: x.get('timestamp', ''), reverse=True) | |
| def execute_query(endpoint, method, data=None): | |
| headers = {"Content-Type": "application/json"} | |
| try: | |
| url = f'{KAPOWARR_BASE_URL}{endpoint}' | |
| params = {'api_key': KAPOWARR_API_TOKEN} | |
| if method == "GET": | |
| response = requests.get(url, headers=headers, params=params) | |
| elif method == "POST": | |
| response = requests.post(url, data=json.dumps(data), headers=headers, params=params) | |
| elif method == "DELETE": | |
| response = requests.delete(url, headers=headers, params=params) | |
| else: | |
| return {"error": "Invalid method"} | |
| if response.status_code == 200: | |
| save_to_log(endpoint, method, data) | |
| return {"success": True, "response": response.json()} | |
| else: | |
| return {"success": False, "response": response.text} | |
| except requests.exceptions.RequestException as e: | |
| return {"error": f"Request failed: {e}"} | |
| def main(): | |
| parser = argparse.ArgumentParser(description="API Query Tool") | |
| parser.add_argument("--endpoint", help="API Endpoint URL") | |
| parser.add_argument("--method", choices=["GET", "POST", "DELETE"], help="HTTP method") | |
| parser.add_argument("--data", help="JSON data for POST method") | |
| args = parser.parse_args() | |
| if not args.endpoint: | |
| args.endpoint = input("Enter the API endpoint: ") | |
| if not args.method: | |
| args.method = input("Enter the HTTP method (GET/POST/DELETE): ").upper() | |
| if args.method == "POST" and not args.data: | |
| args.data = input("Enter the JSON data for POST method: ") | |
| try: | |
| if args.data: | |
| args.data = json.loads(args.data) | |
| except json.JSONDecodeError: | |
| print("Error: Invalid JSON format for data.") | |
| return | |
| result = execute_query(args.endpoint, args.method, args.data) | |
| print(json.dumps(result, indent=2)) | |
| if __name__ == "__main__": | |
| prior_tasks = load_prior_tasks() | |
| print("Loaded Prior Tasks:", prior_tasks) # Add this line for debugging | |
| if prior_tasks: | |
| print("Prior successful tasks:") | |
| for i, task in enumerate(prior_tasks, 1): | |
| timestamp = task.get('timestamp', 'N/A') | |
| endpoint = task.get('endpoint', 'N/A') | |
| method = task.get('method', 'N/A') | |
| data = task.get('data', 'N/A') | |
| # Format for copying to CLI | |
| if endpoint and method: | |
| cli_format = f"./dynamic-api-browse.py --endpoint {endpoint} --method {method}" | |
| if data: | |
| cli_format += f" --data '{json.dumps(data)}'" | |
| print(f"{i}. Timestamp: {timestamp}, Endpoint: {endpoint}, Method: {method}, Data: {data}") | |
| print(f" Copy to CLI: {cli_format}") | |
| main()% |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment