Skip to content

Instantly share code, notes, and snippets.

@Nadiar
Created March 1, 2024 02:50
Show Gist options
  • Select an option

  • Save Nadiar/237aa80908e71b3d93404a63a400c8ef to your computer and use it in GitHub Desktop.

Select an option

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
#!/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