Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.mavera.io/llms.txt

Use this file to discover all available pages before exploring further.

Scenario

Your store uses customer metafields for loyalty tier, preferences, communication preferences, or NPS scores. You pull these via GraphQL, match customers to existing Mavera personas by cohort, and enrich those personas with real-world attributes.

Architecture

Code

import os, requests

STORE = os.environ["SHOPIFY_STORE"]
TOKEN = os.environ["SHOPIFY_ACCESS_TOKEN"]
MV = os.environ["MAVERA_API_KEY"]
SH = f"https://{STORE}.myshopify.com/admin/api/2024-10/graphql.json"
SH_H = {"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"}
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}

QUERY = """query ($cursor: String) {
  customers(first: 100, after: $cursor) {
    edges { node { ordersCount tags metafields(first: 20, namespace: "custom") { edges { node { key value } } } } }
    pageInfo { hasNextPage endCursor }
  }
}"""

customers, cursor = [], None
while True:
    resp = requests.post(SH, json={"query": QUERY, "variables": {"cursor": cursor} if cursor else {}}, headers=SH_H)
    resp.raise_for_status(); data = resp.json()["data"]["customers"]
    for e in data["edges"]:
        n = e["node"]
        mf = {m["node"]["key"]: m["node"]["value"] for m in n["metafields"]["edges"]}
        if mf: customers.append({"orders": n["ordersCount"], "tags": n.get("tags", []), "metafields": mf})
    if not data["pageInfo"]["hasNextPage"]: break
    cursor = data["pageInfo"]["endCursor"]
print(f"{len(customers)} customers with metafields")
def cohort_of(c):
    tags = [t.lower() for t in c.get("tags", [])]
    if c["orders"] >= 6 or "vip" in tags: return "vip"
    if c["orders"] >= 2: return "returning"
    return "new"

personas = requests.get(f"{MB}/personas", headers=MH).json()
personas = personas.get("personas", personas.get("data", []))
all_keys = set(k for c in customers for k in c["metafields"])

for persona in personas:
    text = f"{persona.get('name','')} {persona.get('description','')}".lower()
    cohort = next((c for c in ("vip", "returning", "new") if c in text), None)
    if not cohort: continue
    members = [c for c in customers if cohort_of(c) == cohort]
    if not members: continue
    summary = {}
    for key in all_keys:
        vals = [c["metafields"][key] for c in members if key in c["metafields"]]
        if vals: summary[key] = {"most_common": max(set(vals), key=vals.count), "sample_size": len(vals)}
    resp = requests.patch(f"{MB}/personas/{persona['id']}", json={"demographic": {"custom_attributes": summary}}, headers=MH)
    resp.raise_for_status()
    print(f"Enriched '{persona['name']}' with {len(summary)} attributes")

Example Output

{
  "id": "persona_8f3a1b2c",
  "name": "Shopify VIP Customer",
  "demographic": {
    "custom_attributes": {
      "loyalty_tier": { "most_common": "Platinum", "sample_size": 41 },
      "preferred_category": { "most_common": "Outerwear", "sample_size": 38 },
      "communication_preference": { "most_common": "email", "sample_size": 44 },
      "nps_score": { "most_common": "9", "sample_size": 35 }
    }
  }
}

Error Handling

If the namespace doesn’t match your store’s config, the response returns an empty array without error. Check namespaces in Shopify Admin → Settings → Custom data → Customers. Common namespaces: custom, global, or app-specific like loyalty_app.
If a persona was deleted or the ID is stale, PATCH returns 404. Always fetch the latest list with GET /api/v1/personas before updating. If the persona no longer exists, create a new one with POST.