Delete all the files and vector stores of an OpenAI account

clean_vector_stores.py

#!/usr/bin/env python3
import getpass
import json
import urllib.parse
import urllib.request

def main():
    # 1. Prompt for API key
    api_key = getpass.getpass("Please enter your OpenAI API key: ").strip()
    if not api_key:
        print("No API key provided. Exiting.")
        return

    # We'll use these headers for *all* requests.
    # (The "OpenAI-Beta" header is required for vector store endpoints but doesn't harm for /v1/files.)
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "OpenAI-Beta": "assistants=v2"  # Required for vector store Beta endpoints
    }

    # ----------------------------------------------------------
    # PART 1: VECTOR STORES
    # ----------------------------------------------------------
    vector_stores = list_all_vector_stores(headers)
    print(f"\nFound {len(vector_stores)} vector store(s) in your account.")

    if vector_stores:
        confirm_vs = input("Delete ALL vector stores? [y/N]: ").strip().lower()
        if confirm_vs in ["y", "yes"]:
            for vs in vector_stores:
                vs_id = vs.get("id")
                vs_name = vs.get("name", "")
                print(f"Deleting vector store: {vs_id} (name: '{vs_name}')")
                delete_vector_store(vs_id, headers)
            print("All vector stores deleted.\n")
        else:
            print("No vector stores were deleted.\n")

    # ----------------------------------------------------------
    # PART 2: FILES
    # ----------------------------------------------------------
    files = list_all_files(headers)
    print(f"Found {len(files)} file(s) in your account.")

    if files:
        confirm_files = input("Delete ALL files? [y/N]: ").strip().lower()
        if confirm_files in ["y", "yes"]:
            for f in files:
                file_id = f.get("id")
                file_purpose = f.get("purpose", "")
                print(f"Deleting file: {file_id} (purpose: {file_purpose})")
                delete_file(file_id, headers)
            print("All files deleted.\n")
        else:
            print("No files were deleted.\n")

    print("Done!")

def list_all_vector_stores(headers):
    """
    Fetch all vector stores from the Beta endpoint, handling pagination if necessary.
    Returns a list of vector store objects.
    """
    base_url = "https://api.openai.com/v1/vector_stores"
    all_stores = []
    limit = 20  # up to 100 if you like, but 20 is default

    url = f"{base_url}?limit={limit}"
    while True:
        req = urllib.request.Request(url, headers=headers, method="GET")
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read())

        stores = data.get("data", [])
        all_stores.extend(stores)

        has_more = data.get("has_more", False)
        last_id = data.get("last_id")
        if has_more and last_id:
            # Paginate using ?after=xxx
            parsed_url = list(urllib.parse.urlparse(url))
            query_dict = dict(urllib.parse.parse_qsl(parsed_url[4]))
            query_dict["after"] = last_id
            parsed_url[4] = urllib.parse.urlencode(query_dict)
            url = urllib.parse.urlunparse(parsed_url)
        else:
            break

    return all_stores

def delete_vector_store(vector_store_id, headers):
    """
    DELETE /v1/vector_stores/{vector_store_id}
    """
    url = f"https://api.openai.com/v1/vector_stores/{vector_store_id}"
    req = urllib.request.Request(url, headers=headers, method="DELETE")
    try:
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read())
        if data.get("deleted"):
            print(f"  Successfully deleted: {vector_store_id}")
        else:
            print(f"  Failed to delete: {vector_store_id}")
    except Exception as e:
        print(f"  Error deleting vector store {vector_store_id}: {e}")

def list_all_files(headers):
    """
    Fetch all files from the stable /v1/files endpoint.
    Typically there's no pagination for /v1/files, so this is straightforward.
    Returns a list of file objects.
    """
    url = "https://api.openai.com/v1/files"
    req = urllib.request.Request(url, headers=headers, method="GET")
    try:
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read())
        return data.get("data", [])
    except Exception as e:
        print(f"Error listing files: {e}")
        return []

def delete_file(file_id, headers):
    """
    DELETE /v1/files/{file_id}
    """
    url = f"https://api.openai.com/v1/files/{file_id}"
    req = urllib.request.Request(url, headers=headers, method="DELETE")
    try:
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read())
        if data.get("deleted"):
            print(f"  Successfully deleted: {file_id}")
        else:
            print(f"  Failed to delete: {file_id}")
    except Exception as e:
        print(f"  Error deleting file {file_id}: {e}")

if __name__ == "__main__":
    main()