s3.py

1import boto32import json3import time4from catalog.config import CARLOMITCHENER_BUCKET, DELAY5from catalog.helpers import product_path6from env import load_json78s3 = boto3.client("s3")910PREFIX = "data/products/"1112def product_key(id: int) -> str:13    return f"{PREFIX}{id}.json"1415def put_json(key: str, data) -> None:16    s3.put_object(17        Bucket=CARLOMITCHENER_BUCKET,18        Key=key,19        Body=json.dumps(data),20        ContentType="application/json",21    )2223def upload_products(ids: list[int]) -> None:24    print(f"Uploading {len(ids)} products to {CARLOMITCHENER_BUCKET}/{PREFIX}")25    for id in ids:26        key = product_key(id)27        put_json(key, load_json(product_path(id)))28        print(f"Uploaded: {key}")29        time.sleep(DELAY)30    print(f"Uploaded {len(ids)} products")3132def list_products() -> list[str]:33    keys = []34    paginator = s3.get_paginator("list_objects_v2")35    for page in paginator.paginate(Bucket=CARLOMITCHENER_BUCKET, Prefix=PREFIX):36        keys.extend(obj["Key"] for obj in page.get("Contents", []))37    return keys3839def delete_products() -> None:40    keys = list_products()41    if not keys:42        print(f"Nothing under {CARLOMITCHENER_BUCKET}/{PREFIX}")43        return44    for i in range(0, len(keys), 1000):45        batch = [{"Key": key} for key in keys[i:i + 1000]]46        s3.delete_objects(Bucket=CARLOMITCHENER_BUCKET, Delete={"Objects": batch})47    print(f"Deleted {len(keys)} objects under {CARLOMITCHENER_BUCKET}/{PREFIX}")