status.py

1from automator.core.api import shopify_request2from automator.core.errors import Retry, TaskAborted3from automator.core.models import Task4from automator.core.steps import Step56MAX_STATUS_ATTEMPTS = 1078QUERY = """9query checkFiles($ids: [ID!]!) {10    nodes(ids: $ids) {11        ... on MediaImage {12            alt13            fileStatus14            id15        }16    }17}18"""1920def get_status(task: Task) -> list[dict]:21    variables = {"ids": [m.shopify_id for m in task.mockups]}22    result = shopify_request(task, QUERY, variables)23    return [node for node in result["data"]["nodes"] if node]2425def retry_files(task: Task) -> None:26    if task.metadata.get("files_retried"):27        raise TaskAborted(f"shopify file FAILED twice for {task.key}")28    task.metadata["files_retried"] = True29    for mockup in task.mockups:30        mockup.shopify_id = None31    task.place(Step.FILES)32    raise Retry(f"Current: {Step.STATUS}. Next: {Step.FILES}")3334def check_status(task: Task, files: list[dict]) -> None:35    if any(file["fileStatus"] == "FAILED" for file in files):36        retry_files(task)37    if not all(file["fileStatus"] == "READY" for file in files):38        count = task.metadata.get("status_count", 0) + 139        task.metadata["status_count"] = count40        if count >= MAX_STATUS_ATTEMPTS:41            raise TaskAborted(f"status_count {count} for task {task.key}")42        raise Retry(f"Current: {Step.STATUS}. Next: {Step.STATUS} ({count}/{MAX_STATUS_ATTEMPTS})")4344def mrly_status(task: Task) -> Task:45    files = get_status(task)46    if len(files) != len(task.mockups):47        raise TaskAborted(f"shopify returned {len(files)} of {len(task.mockups)} files")48    check_status(task, files)49    task.place(Step.PRODUCT)50    return task