I could do with a button that deletes 100 test robot.
#!/usr/bin/env python3
"""
delete_test_robots.py
Delete up to 100 test‑robots from a Dreamgrid deployment via its REST API.
Author:
Date: 2026‑01‑25
"""
import sys
import time
import json
import logging
from typing import List, Dict, Any, Optional
import requests
# ----------------------------------------------------------------------
# ── CONFIGURATION ───────────────────────────────────────────────────────
# ----------------------------------------------------------------------
# 1️⃣ Base URL of the Dreamgrid API (no trailing slash!)
API_BASE_URL = "
https://api.dreamgrid.example.com/v1"
# 2️⃣ Authentication token / API key.
# If the API uses a simple Bearer token, put it here:
API_BEARER_TOKEN = "YOUR_BEARER_TOKEN_HERE"
# 3️⃣ How to identify a *test* robot.
# Adjust ONE of the following helper functions to match your schema.
#
# • If robots have a `name` field that contains "test" → use `name_contains_test`.
# • If they have a boolean `is_test` flag → use `is_test_flag`.
# • If they are tagged with a specific tag → use `has_tag`.
#
# 4️⃣ Pagination settings (depends on your API).
PAGE_SIZE = 100 # Number of items the API returns per page.
MAX_DELETE = 100 # Upper bound on deletions performed by this script.
# ----------------------------------------------------------------------
# ── LOGGING SETUP ───────────────────────────────────────────────────────
# ----------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
handlers=[logging.StreamHandler(sys.stdout)]
)
log = logging.getLogger(__name__)
# ----------------------------------------------------------------------
# ── AUTH HELPERS ───────────────────────────────────────────────────────
# ----------------------------------------------------------------------
def get_auth_headers() -> Dict[str, str]:
"""
Return the HTTP headers needed for authenticated requests.
Modify this function if your API uses a different auth method.
"""
if not API_BEARER_TOKEN or API_BEARER_TOKEN == "YOUR_BEARER_TOKEN_HERE":
raise RuntimeError("Please set a valid API_BEARER_TOKEN in the script.")
return {
"Authorization": f"Bearer {API_BEARER_TOKEN}",
"Accept": "application/json",
"Content-Type": "application/json"
}
# ----------------------------------------------------------------------
# ── FILTER HELPERS ───────────────────────────────────────────────────────
# ----------------------------------------------------------------------
def name_contains_test(robot: Dict[str, Any]) -> bool:
"""
Example filter: robot's `name` field contains the word "test" (case‑insensitive).
Adjust the field name if your API uses a different key.
"""
name = robot.get("name", "")
return "test" in name.lower()
def is_test_flag(robot: Dict[str, Any]) -> bool:
"""If the robot JSON has a boolean `is_test` flag."""
return robot.get("is_test") is True
def has_tag(robot: Dict[str, Any], tag: str = "test-robot") -> bool:
"""
If robots have a list of tags, check for a specific tag.
Example: robot["tags"] == ["alpha", "test-robot"]
"""
tags = robot.get("tags", [])
return tag in tags
# Pick the filter you need (uncomment the appropriate line):
SELECTOR = name_contains_test
# SELECTOR = is_test_flag
# SELECTOR = lambda r: has_tag(r, tag="test-robot")
# ----------------------------------------------------------------------
# ── API CALLS ───────────────────────────────────────────────────────────
# ----------------------------------------------------------------------
def fetch_robots(page: int = 1, page_size: int = PAGE_SIZE) -> List[Dict[str, Any]]:
"""
Retrieve a page of robot objects from Dreamgrid.
Returns a list of robot dicts (empty list → no more data).
"""
url = f"{API_BASE_URL}/robots"
params = {"page": page, "page_size": page_size}
response = requests.get(url, headers=get_auth_headers(), params=params, timeout=15)
if response.status_code != 200:
raise RuntimeError(f"Failed to fetch robots (HTTP {response.status_code}): {response.text}")
data = response.json()
# Adjust according to your API's envelope format.
# Many APIs return {"results": [...], "total": 123, "page": 1, ...}
robots = data.get("results") or data.get("robots") or data
if not isinstance(robots, list):
raise RuntimeError("Unexpected response format – could not locate robot list.")
return robots
def delete_robot(robot_id: str) -> bool:
"""
Issue a DELETE request for a single robot.
Returns True on success, False otherwise.
"""
url = f"{API_BASE_URL}/robots/{robot_id}"
response = requests.delete(url, headers=get_auth_headers(), timeout=15)
if response.status_code in (200, 204):
return True
else:
log.warning(f"Delete failed for robot {robot_id}: HTTP {response.status_code} – {response.text}")
return False
# ----------------------------------------------------------------------
# ── MAIN LOGIC ────────────────────────────────────────────────────────
# ----------------------------------------------------------------------
def main() -> None:
log.info("=== Dreamgrid Test‑Robot Cleanup Script ===")
log.info(f"Target: delete up to {MAX_DELETE} test robots.")
log.info("Fetching robot list...")
deleted_ids: List[str] = []
total_found = 0
page = 1
while len(deleted_ids) < MAX_DELETE:
try:
robots = fetch_robots(page=page)
except Exception as e:
log.error(f"Unable to fetch page {page}: {e}")
break
if not robots: # No more data
log.info("No more robots returned by the API – stopping.")
break
# Filter the page for test robots
test_robots = [r for r in robots if SELECTOR(r)]
if not test_robots:
log.debug(f"Page {page} contains no test robots.")
else:
log.info(f"Page {page}: found {len(test_robots)} test robot(s).")
for robot in test_robots:
if len(deleted_ids) >= MAX_DELETE:
break
robot_id = str(robot.get("id"))
if not robot_id:
log.warning("Encountered robot entry without an `id` field – skipping.")
continue
total_found += 1
log.info(f"Deleting robot #{total_found} (ID: {robot_id}) …")
try:
success = delete_robot(robot_id)
except Exception as exc:
log.error(f"Exception while deleting robot {robot_id}: {exc}")
success = False
if success:
deleted_ids.append(robot_id)
log.info(f"✅ Deleted robot {robot_id}. ({len(deleted_ids)}/{MAX_DELETE})")
else:
log.warning(f"❌ Failed to delete robot {robot_id}. Continuing…")
# Be nice to the server – a short pause helps avoid rate‑limit hits.
time.sleep(0.1)
# Move to the next page (only if we haven't exhausted the deletion quota)
page += 1
log.info("=== Summary ===")
log.info(f"Total test robots examined: {total_found}")
log.info(f"Successfully deleted: {len(deleted_ids)}")
if deleted_ids:
log.info(f"Deleted IDs: {', '.join(deleted_ids)}")
else:
log.info("No robots were deleted. Verify your filter logic or that test robots exist.")
log.info("=== Done ===")
# ----------------------------------------------------------------------
# ── ENTRY POINT ───────────────────────────────────────────────────────
# ----------------------------------------------------------------------
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
log.warning("Interrupted by user – exiting.")
sys.exit(1)
except Exception as e:
log.exception(f"Unexpected error: {e}")
sys.exit(1)