Datastore bulk delete

datastore_bulk_delete.py

#!/usr/bin/env python
import re
import sys
from argparse import ArgumentParser

from google.cloud import datastore


def get_client(project: str, database: str | None) -> datastore.Client:
    return datastore.Client(project=project, database=database)


def _delete_keys(client: datastore.Client, keys):
    for batch in (keys[i : i + 100] for i in range(0, len(keys), 100)):
        client.delete_multi(batch)


def parse_filters(filters: list[str]) -> list[tuple[str, str, str]]:
    result = []
    for f in filters:
        match = re.match("^(\w+)\s(=|!=|<|<=|>|>=|in|not in)\s(\w+)", f)
        if not match:
            raise ValueError(f"Invalid filter '{f}'")
        key = match.group(1)
        operator = match.group(2)
        value = match.group(3)
        if value.lower() == "true":
            value = True
        elif value.lower() == "false":
            value = False
        result.append((key, operator, value))
    return result


def wipe_data(client: datastore.Client, kind: str, filters: list[tuple[str, str, str]]) -> None:
    """Delete all entities across all stores for a given SKU"""
    # Products
    query = client.query(kind=kind)
    for f in filters:
        query.add_filter(*f)
    print(f"Filters: {query.filters}")
    keys = [row.key for row in query.fetch()]
    confirm = input(
        f"Found {len(keys)} results matching query.\nAre you sure you want to wipe all data? [y/n] "
    )
    if not confirm.lower().strip().startswith("y"):
        sys.exit(0)

    _delete_keys(client, keys)


if __name__ == "__main__":
    parser = ArgumentParser()
    parser.add_argument("--project", help="Gcloud project name", required=True)
    parser.add_argument("--database", help="Database to delete products from", default=None)
    parser.add_argument("--kind", help="Database to delete products from", required=True)
    parser.add_argument("-f", "--filters", action="append", help="Query filters", default=[])
    args = parser.parse_args()
    datastore_client = get_client(args.project, args.database)
    wipe_data(datastore_client, kind=args.kind, filters=parse_filters(args.filters))