import boto3
REGION = "us-west-1"
TABLE_NAME = "S3-object-size-history"
def empty_dynamodb_table(table_name: str, region: str) -> int:
dynamodb = boto3.resource("dynamodb", region_name=region)
table = dynamodb.Table(table_name)
# Get primary key attributes (PK / SK)
key_names = [k["AttributeName"] for k in table.key_schema]
# Only scan key attributes (cheaper)
expr_attr_names = {f"#{k}": k for k in key_names}
proj_expr = ", ".join(expr_attr_names.keys())
deleted = 0
scan_kwargs = {
"ProjectionExpression": proj_expr,
"ExpressionAttributeNames": expr_attr_names,
}
while True:
resp = table.scan(**scan_kwargs)
items = resp.get("Items", [])
with table.batch_writer() as batch:
for it in items:
key = {k: it[k] for k in key_names}
batch.delete_item(Key=key)
deleted += 1
last_key = resp.get("LastEvaluatedKey")
if not last_key:
break
scan_kwargs["ExclusiveStartKey"] = last_key
return deleted
if __name__ == "__main__":
n = empty_dynamodb_table(TABLE_NAME, REGION)
print(f"Deleted {n} items from {TABLE_NAME}")