Pick the packing insert from the quiz result

Goal: the box of a shopper who took the quiz carries the insert for their routine (a “dry skin: how to use this” card, a sample), chosen by the quiz result, without anyone at the packing station looking it up.

Uses: People (GET /v1/profiles, GET /v1/profiles/{profile_id}, key with profiles:read) on your server; Shopify’s Admin API (orders, metafieldsSet, tagsAdd) with read_orders and write_orders. The result reaches the packing app as an order tag (insert:dry-routine) and an order metafield (octane.packing_insert); use whichever your packing app reads (ShipStation, Shippo and most WMS apps filter on tags; some read metafields).

Two ways to know the result

1. On the order. When the quiz’s result rides to the cart (Put the quiz result on the order), every order from that cart carries quiz_result in customAttributes. No lookup, exact, and it works for guests.
2. On the person. When the attribute is missing (an accelerated checkout dropped it, the quiz was taken on another device, an order placed days later), the order’s email finds the person, and latest_results[] names the result page they reached.

The script below does 1 first and falls back to 2. Run it every few minutes (a cron, a scheduled job), or call pick_insert from your own orders/create webhook handler; either way it only touches orders that carry no insert: tag yet.

No code at all

If the attribute is always there (way 1), Shopify Flow does this without a server: trigger Order created, condition order.customAttributes has a key quiz_result, action Add order tags with insert:{{ order.customAttributes.quiz_result }}. The script is for the fallback to the person, and for stores without Flow.

The script

Python

import os
import httpx

OCTANE = {"Authorization": f"Bearer {os.environ['OCTANE_API_KEY']}"}
QUIZ_ID = "quiz_7c9e6679742540de944be07fc1f90ae7"
ADMIN_URL = f"https://{os.environ['SHOPIFY_SHOP']}/admin/api/2025-07/graphql.json"
ADMIN_HEADERS = {"X-Shopify-Access-Token": os.environ["SHOPIFY_ADMIN_TOKEN"]}
INSERTS = {"result-dry": "dry-routine", "result-oily": "oily-routine", "result-combination": "combination-routine"}  # result page key -> insert

def admin(query: str, variables: dict) -> dict:
    r = httpx.post(ADMIN_URL, headers=ADMIN_HEADERS, json={"query": query, "variables": variables}, timeout=30)
    r.raise_for_status()
    body = r.json()
    if body.get("errors"):
        raise RuntimeError(body["errors"])
    return body["data"]

def result_from_person(email: str) -> str | None:
    r = httpx.get("https://api.octaneai.com/v1/profiles", params={"q": email, "quiz_id": QUIZ_ID}, headers=OCTANE, timeout=30)
    r.raise_for_status()
    rows = r.json()["data"]
    if not rows:
        return None
    r = httpx.get(f"https://api.octaneai.com/v1/profiles/{rows[0]['id']}", headers=OCTANE, timeout=30)
    r.raise_for_status()
    result = next((x for x in r.json()["latest_results"] if x["quiz_id"] == QUIZ_ID), None)
    return result and result["result_page"] and result["result_page"]["page_key"]

def pick_insert(order: dict) -> str | None:
    attributes = {a["key"]: a["value"] for a in order["customAttributes"]}
    result = attributes.get("quiz_result") or (order["email"] and result_from_person(order["email"]))
    return INSERTS.get(result or "")

def tag_order(order_id: str, insert: str) -> None:
    admin(
        """mutation ($id: ID!, $tags: [String!]!, $metafields: [MetafieldsSetInput!]!) {
             tagsAdd(id: $id, tags: $tags) { userErrors { field message } }
             metafieldsSet(metafields: $metafields) { userErrors { field message } } }""",
        {"id": order_id, "tags": [f"insert:{insert}"],
         "metafields": [{"ownerId": order_id, "namespace": "octane", "key": "packing_insert", "type": "single_line_text_field", "value": insert}]},
    )

def run(since: str) -> None:
    data = admin(
        """query ($q: String!) { orders(first: 50, query: $q, sortKey: CREATED_AT) {
             nodes { id name email tags customAttributes { key value } } } }""",
        {"q": f"created_at:>='{since}' -tag:insert:*"},
    )
    for order in data["orders"]["nodes"]:
        if any(tag.startswith("insert:") for tag in order["tags"]):
            continue  # tagged by a run the search index has not caught up with
        insert = pick_insert(order)
        if insert:
            tag_order(order["id"], insert)
        print(order["name"], insert)  # #1042 dry-routine

if __name__ == "__main__":
    run("2026-09-07T00:00:00Z")

TypeScript

const OCTANE = { Authorization: `Bearer ${process.env.OCTANE_API_KEY}` };
const QUIZ_ID = "quiz_7c9e6679742540de944be07fc1f90ae7";
const ADMIN_URL = `https://${process.env.SHOPIFY_SHOP}/admin/api/2025-07/graphql.json`;
const ADMIN_HEADERS = { "X-Shopify-Access-Token": process.env.SHOPIFY_ADMIN_TOKEN!, "Content-Type": "application/json" };
const INSERTS: Record<string, string> = { "result-dry": "dry-routine", "result-oily": "oily-routine", "result-combination": "combination-routine" }; // result page key -> insert

type Order = { id: string; name: string; email: string | null; tags: string[]; customAttributes: { key: string; value: string }[] };

async function admin(query: string, variables: Record<string, unknown>) {
  const res = await fetch(ADMIN_URL, { method: "POST", headers: ADMIN_HEADERS, body: JSON.stringify({ query, variables }) });
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  const body = await res.json();
  if (body.errors) throw new Error(JSON.stringify(body.errors));
  return body.data;
}

async function resultFromPerson(email: string): Promise<string | null> {
  const qs = new URLSearchParams({ q: email, quiz_id: QUIZ_ID });
  let res = await fetch(`https://api.octaneai.com/v1/profiles?${qs}`, { headers: OCTANE });
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  const rows = (await res.json()).data;
  if (!rows.length) return null;
  res = await fetch(`https://api.octaneai.com/v1/profiles/${rows[0].id}`, { headers: OCTANE });
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  const result = (await res.json()).latest_results.find((r: { quiz_id: string }) => r.quiz_id === QUIZ_ID);
  return result?.result_page?.page_key ?? null;
}

async function pickInsert(order: Order): Promise<string | null> {
  const attribute = order.customAttributes.find((a) => a.key === "quiz_result")?.value;
  const result = attribute || (order.email ? await resultFromPerson(order.email) : null);
  return result ? INSERTS[result] ?? null : null;
}

async function tagOrder(orderId: string, insert: string) {
  await admin(
    `mutation ($id: ID!, $tags: [String!]!, $metafields: [MetafieldsSetInput!]!) {
       tagsAdd(id: $id, tags: $tags) { userErrors { field message } }
       metafieldsSet(metafields: $metafields) { userErrors { field message } } }`,
    { id: orderId, tags: [`insert:${insert}`],
      metafields: [{ ownerId: orderId, namespace: "octane", key: "packing_insert", type: "single_line_text_field", value: insert }] },
  );
}

async function run(since: string) {
  const data = await admin(
    `query ($q: String!) { orders(first: 50, query: $q, sortKey: CREATED_AT) {
       nodes { id name email tags customAttributes { key value } } } }`,
    { q: `created_at:>='${since}' -tag:insert:*` },
  );
  for (const order of data.orders.nodes as Order[]) {
    if (order.tags.some((tag) => tag.startsWith("insert:"))) continue; // tagged by a run the search index has not caught up with
    const insert = await pickInsert(order);
    if (insert) await tagOrder(order.id, insert);
    console.log(order.name, insert); // #1042 dry-routine
  }
}

run("2026-09-07T00:00:00Z");

Go

package main

import (
  "bytes"
  "encoding/json"
  "fmt"
  "log"
  "net/http"
  "net/url"
  "os"
  "slices"
  "strings"
)

const quizID = "quiz_7c9e6679742540de944be07fc1f90ae7"

var inserts = map[string]string{"result-dry": "dry-routine", "result-oily": "oily-routine", "result-combination": "combination-routine"} // result page key -> insert

type order struct {
  ID               string  `json:"id"`
  Name             string  `json:"name"`
  Email            *string `json:"email"`
  Tags             []string `json:"tags"`
  CustomAttributes []struct {
    Key   string `json:"key"`
    Value string `json:"value"`
  } `json:"customAttributes"`
}

func admin(query string, variables map[string]any, into any) {
  body, _ := json.Marshal(map[string]any{"query": query, "variables": variables})
  req, _ := http.NewRequest("POST", "https://"+os.Getenv("SHOPIFY_SHOP")+"/admin/api/2025-07/graphql.json", bytes.NewReader(body))
  req.Header.Set("X-Shopify-Access-Token", os.Getenv("SHOPIFY_ADMIN_TOKEN"))
  req.Header.Set("Content-Type", "application/json")
  res, err := http.DefaultClient.Do(req)
  if err != nil {
    log.Fatal(err)
  }
  defer res.Body.Close()
  var out struct {
    Data   json.RawMessage `json:"data"`
    Errors []any           `json:"errors"`
  }
  json.NewDecoder(res.Body).Decode(&out)
  if len(out.Errors) > 0 {
    log.Fatal(out.Errors)
  }
  if into != nil {
    json.Unmarshal(out.Data, into)
  }
}

func octane(path string, into any) bool {
  req, _ := http.NewRequest("GET", "https://api.octaneai.com"+path, nil)
  req.Header.Set("Authorization", "Bearer "+os.Getenv("OCTANE_API_KEY"))
  res, err := http.DefaultClient.Do(req)
  if err != nil {
    log.Fatal(err)
  }
  defer res.Body.Close()
  if res.StatusCode != 200 {
    log.Fatalf("status %d", res.StatusCode)
  }
  json.NewDecoder(res.Body).Decode(into)
  return true
}

func resultFromPerson(email string) string {
  var list struct {
    Data []struct {
      ID string `json:"id"`
    } `json:"data"`
  }
  q := url.Values{"q": {email}, "quiz_id": {quizID}}
  octane("/v1/profiles?"+q.Encode(), &list)
  if len(list.Data) == 0 {
    return ""
  }
  var profile struct {
    LatestResults []struct {
      QuizID     string `json:"quiz_id"`
      ResultPage *struct {
        PageKey string `json:"page_key"`
      } `json:"result_page"`
    } `json:"latest_results"`
  }
  octane("/v1/profiles/"+list.Data[0].ID, &profile)
  for _, r := range profile.LatestResults {
    if r.QuizID == quizID && r.ResultPage != nil {
      return r.ResultPage.PageKey
    }
  }
  return ""
}

func pickInsert(o order) string {
  result := ""
  for _, a := range o.CustomAttributes {
    if a.Key == "quiz_result" {
      result = a.Value
    }
  }
  if result == "" && o.Email != nil {
    result = resultFromPerson(*o.Email)
  }
  return inserts[result]
}

func main() {
  var data struct {
    Orders struct {
      Nodes []order `json:"nodes"`
    } `json:"orders"`
  }
  admin(`query ($q: String!) { orders(first: 50, query: $q, sortKey: CREATED_AT) {
           nodes { id name email tags customAttributes { key value } } } }`,
    map[string]any{"q": "created_at:>='2026-09-07T00:00:00Z' -tag:insert:*"}, &data)
  for _, o := range data.Orders.Nodes {
    if slices.ContainsFunc(o.Tags, func(t string) bool { return strings.HasPrefix(t, "insert:") }) {
      continue // tagged by a run the search index has not caught up with
    }
    insert := pickInsert(o)
    if insert != "" {
      admin(`mutation ($id: ID!, $tags: [String!]!, $metafields: [MetafieldsSetInput!]!) {
               tagsAdd(id: $id, tags: $tags) { userErrors { field message } }
               metafieldsSet(metafields: $metafields) { userErrors { field message } } }`,
        map[string]any{"id": o.ID, "tags": []string{"insert:" + insert}, "metafields": []map[string]string{{
          "ownerId": o.ID, "namespace": "octane", "key": "packing_insert", "type": "single_line_text_field", "value": insert}}}, nil)
    }
    fmt.Println(o.Name, insert) // #1042 dry-routine
  }
}

result_page.page_key is the key of the result page the person reached on their latest completed run of the quiz, the same value terminal_page carries on the storefront and on the quiz.finished webhook, so INSERTS is keyed once whichever way the result arrives. The map is yours: a result page per routine, an insert per result page.

Errors to handle

Where	What	Do
GET /v1/profiles	403 insufficient_scope	The key needs profiles:read.
GET /v1/profiles	404 not_found	quiz_id is not this store’s or the quiz is deleted.
GET /v1/profiles	429 rate_limited	Wait Retry-After; the batch is not lost, the next run picks the orders up (they still carry no insert: tag).
GET /v1/profiles/{profile_id}	404 not_found	The person asked to be removed between the two calls; ship without an insert.
orders query	errors with ACCESS_DENIED	The app needs read_orders (read_all_orders for orders older than 60 days).
tagsAdd / metafieldsSet	userErrors	The order is archived or the metafield key is malformed; log and skip, never retry in a loop.

Things to know

• The tag is the marker that an order has been handled. Shopify’s order search runs on an index that trails the store by up to a minute: a just-placed order is not in the results yet, and an order tagged seconds ago still comes back, which is why the script also skips any row that already carries an insert: tag. Run it on a schedule with a window wider than the interval (every 5 minutes over the last hour) and every order is picked up once. An order that maps to no insert stays untagged and is looked at on every run; tag those insert:none if the list grows.
• orders with sortKey: CREATED_AT and a created_at:>= window returns at most 50 here; page with pageInfo { hasNextPage endCursor } when a run can see more than that.
• A person has one latest_results entry per quiz, their most recent completed run. A shopper who retook the quiz after ordering gets the newer result; run the script soon after the order.
• The shopper’s email goes to GET /v1/profiles?q=, nowhere else; the insert name is the only thing written to the order. Do not copy answers onto the order metafield: every staff account and every app with read_orders reads it.
• Packing apps import tags at order sync. Run this before the packing app syncs (a few minutes after the order) or trigger their re-sync; a tag added after the shipment is printed changes nothing.