s3.py

1import boto32import json3import time4from botocore.exceptions import ClientError5from .api import delete_product, logger6from .config import CARLOMITCHENER_BUCKET, SITE_URL7from .errors import NoTaskError8from .models import Task910s3 = boto3.client("s3")1112BUCKET = CARLOMITCHENER_BUCKET13AUTOMATOR_KEY = "data/automator.json"14PATHS_KEY = "data/paths.json"15SITE_PREFIX = "site/"16CDN_PREFIX = f"{SITE_PREFIX}cdn/printful/"1718# KEYS1920def product_key(id: int) -> str:21    return f"data/products/{id}.json"2223def task_key(key: str) -> str:24    return f"data/tasks/{key}/{key}.json"2526def task_prefix(key: str) -> str:27    return f"data/tasks/{key}/"2829def cdn_prefix(key: str) -> str:30    return f"{CDN_PREFIX}{key}/"3132def cdn_key(key: str, name: str) -> str:33    return f"{CDN_PREFIX}{key}/{name}.png"3435def s3_url(key: str) -> str:36    if key.startswith(SITE_PREFIX):37        key = key[len(SITE_PREFIX):]38    return f"{SITE_URL}/{key}"3940# OBJECTS4142def put_json(key: str, data) -> None:43    s3.put_object(44        Bucket=BUCKET,45        Key=key,46        Body=json.dumps(data),47        ContentType="application/json",48    )4950def get_json(key: str) -> dict:51    try:52        response = s3.get_object(Bucket=BUCKET, Key=key)53    except ClientError as error:54        if error.response["Error"]["Code"] in ("NoSuchKey", "404"):55            raise NoTaskError(key)56        raise57    return json.loads(response["Body"].read().decode("utf-8"))5859def put_png(key: str, data) -> None:60    s3.upload_fileobj(data, Bucket=BUCKET, Key=key, ExtraArgs={"ContentType": "image/png"})6162def key_exists(prefix: str) -> bool:63    response = s3.list_objects_v2(Bucket=BUCKET, Prefix=prefix, MaxKeys=1)64    return "Contents" in response6566def list_objects(prefix: str) -> list[dict]:67    results = []68    paginator = s3.get_paginator("list_objects_v2")69    for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):70        results.extend(page.get("Contents", []))71    return results7273def delete_folder(prefix: str) -> int:74    objects = list_objects(prefix)75    for i in range(0, len(objects), 1000):76        batch = [{"Key": obj["Key"]} for obj in objects[i:i + 1000]]77        s3.delete_objects(Bucket=BUCKET, Delete={"Objects": batch})78    return len(objects)7980# CATALOG8182def task_exists(key: str) -> bool:83    return key_exists(task_key(key))8485def load_product(id: int) -> dict:86    return get_json(product_key(id))8788def load_paths() -> dict:89    return get_json(PATHS_KEY)9091def save_paths(data: dict) -> None:92    put_json(PATHS_KEY, data)9394# TASK9596def load_task() -> Task:97    data = get_json(AUTOMATOR_KEY)98    if not data:99        raise NoTaskError(AUTOMATOR_KEY)100    return Task.from_dict(data)101102def save_task(task: Task) -> None:103    task.updated_at = int(time.time())104    put_json(AUTOMATOR_KEY, task.to_dict())105106def archive_task(task: Task) -> None:107    put_json(task_key(task.key), task.to_dict())108109def clear_task() -> None:110    put_json(AUTOMATOR_KEY, {})111112def abort_task(task: Task) -> None:113    if task.product.shopify_id:114        try:115            delete_product(task)116        except Exception as error:117            logger.info(f"{task.desc} productDelete failed during abort: {error}")118    delete_folder(task_prefix(task.key))119    delete_folder(cdn_prefix(task.key))120    paths = load_paths()121    paths[str(task.product.id)] = None122    save_paths(paths)123    clear_task()124    logger.info(f"{task.desc} quarantined product {task.product.id}")