api.py

1import json2from mrlypy.core.helpers import create_logger3from . import http4from .config import (5    PRINTFUL_API_KEY,6    PRINTFUL_URL,7    SHOPIFY_ADMIN_URL,8    SHOPIFY_AUTH_URL,9    SHOPIFY_CLIENT_ID,10    SHOPIFY_SECRET,11)12from .errors import PrintfulError, Retry, ShopifyError, TaskAborted13from .models import Task1415BODY_LIMIT = 3001617logger = create_logger("carlomitchener-automator")1819TOKEN = {"admin": None}2021def extract_gid(gid: str) -> str:22    return gid.split("/")[-1]2324def clip(text: str) -> str:25    return text[:BODY_LIMIT].replace("\n", " ")2627def log(task: Task, method: str, url: str, status: int) -> None:28    logger.info(f"{task.desc} {task.step} {method} {url} {status}")2930# TOKEN3132def mint_token() -> None:33    payload = {34        "client_id": SHOPIFY_CLIENT_ID,35        "client_secret": SHOPIFY_SECRET,36        "grant_type": "client_credentials",37    }38    response = http.request("POST", SHOPIFY_AUTH_URL, data=payload)39    if response.status_code != 200:40        raise ShopifyError(f"admin token {response.status_code}")41    result = response.json()42    if "access_token" not in result:43        raise ShopifyError("admin token refused")44    TOKEN["admin"] = result["access_token"]45    logger.info(f"admin token minted for {SHOPIFY_ADMIN_URL}")4647# PRINTFUL4849def printful_request(task: Task, method: str, path: str, data: dict = None, allow_404: bool = False) -> dict:50    url = f"{PRINTFUL_URL}{path}"51    headers = {"Authorization": f"Bearer {PRINTFUL_API_KEY}"}52    response = http.request(method, url, headers=headers, json_data=data)53    log(task, method, url, response.status_code)54    if response.status_code == 429:55        raise Retry("printful 429")56    if response.status_code == 400:57        raise TaskAborted(f"printful 400 {clip(response.text)}")58    if response.status_code == 404:59        if allow_404:60            return {}61        raise TaskAborted(f"printful 404 {clip(response.text)}")62    if response.status_code != 200:63        raise PrintfulError(f"printful {response.status_code} {clip(response.text)}")64    return response.json()6566# SHOPIFY6768def shopify_request(task: Task, query: str, variables: dict = None) -> dict:69    headers = {"X-Shopify-Access-Token": TOKEN["admin"]}70    payload = {"query": query, "variables": variables}71    response = http.request("POST", SHOPIFY_ADMIN_URL, headers=headers, json_data=payload)72    log(task, "POST", SHOPIFY_ADMIN_URL, response.status_code)73    if response.status_code == 429:74        raise Retry("shopify 429")75    if response.status_code != 200:76        raise ShopifyError(f"shopify {response.status_code}")77    result = response.json()78    if "errors" in result:79        raise ShopifyError(f"shopify errors {clip(json.dumps(result['errors']))}")80    return result8182def check_errors(result: dict, name: str) -> dict:83    data = result["data"][name]84    errors = data["userErrors"]85    if errors:86        raise TaskAborted(f"{name} userErrors {clip(json.dumps(errors))}")87    return data8889DELETE_PRODUCT = """90mutation productDelete($input: ProductDeleteInput!) {91    productDelete(input: $input) {92        deletedProductId93        userErrors {94            field95            message96        }97    }98}99"""100101def delete_product(task: Task) -> None:102    variables = {"input": {"id": task.product.shopify_id}}103    result = shopify_request(task, DELETE_PRODUCT, variables)104    check_errors(result, "productDelete")105    logger.info(f"{task.desc} deleted shopify product {task.product.shopify_id}")