How to Automate Shopify Abandoned Cart Recovery using the API

1import os, hmac, hashlib, base64, logging, requests
2from flask import Flask, request, abort
3from celery import Celery
4
5logging.basicConfig(level=logging.INFO)
6app = Flask(__name__)
7celery = Celery('cart', broker=os.environ['REDIS_URL'])
8
9SHOP_DOMAIN = os.environ['SHOPIFY_SHOP_DOMAIN']
10ACCESS_TOKEN = os.environ['SHOPIFY_ACCESS_TOKEN']
11APP_SECRET = os.environ['SHOPIFY_APP_SECRET']
12SENDGRID_KEY = os.environ['SENDGRID_API_KEY']
13TEMPLATE_FIRST = os.environ['SENDGRID_CART_TEMPLATE_1']
14TEMPLATE_SECOND = os.environ['SENDGRID_CART_TEMPLATE_2']
15
16GQL_URL = f'https://{SHOP_DOMAIN}/admin/api/2026-04/graphql.json'
17GQL_HEADERS = {'X-Shopify-Access-Token': ACCESS_TOKEN, 'Content-Type': 'application/json'}
18
19def gql(query, variables=None):
20    resp = requests.post(GQL_URL, json={'query': query, 'variables': variables or {}}, headers=GQL_HEADERS)
21    resp.raise_for_status()
22    return resp.json()
23
24def find_abandoned_checkout(token):
25    q = '{ abandonedCheckouts(first: 20) { edges { node { id abandonedCheckoutUrl email totalPriceV2 { amount currencyCode } lineItems(first: 10) { edges { node { title quantity } } } } } } }'
26    checkouts = gql(q)['data']['abandonedCheckouts']['edges']
27    return next((e['node'] for e in checkouts if token in e['node'].get('abandonedCheckoutUrl', '')), None)
28
29def send_email(to_email, template_id, data):
30    payload = {'from': {'email': 'hello@yourstore.com'}, 'personalizations': [{'to': [{'email': to_email}], 'dynamic_template_data': data}], 'template_id': template_id}
31    resp = requests.post('https://api.sendgrid.com/v3/mail/send', json=payload, headers={'Authorization': f'Bearer {SENDGRID_KEY}'})
32    return resp.status_code == 202
33
34@celery.task
35def first_recovery_check(token, email):
36    checkout = find_abandoned_checkout(token)
37    if not checkout:
38        logging.info(f'Cart {token} completed, no recovery needed')
39        return
40    items = [{'title': e['node']['title'], 'qty': e['node']['quantity']} for e in checkout['lineItems']['edges']]
41    data = {'recovery_url': checkout['abandonedCheckoutUrl'], 'cart_total': checkout['totalPriceV2']['amount'], 'items': items}
42    if send_email(email, TEMPLATE_FIRST, data):
43        logging.info(f'First recovery email sent to {email}')
44        second_recovery_check.apply_async(args=[token, email], countdown=86400)  # 24h
45
46@celery.task
47def second_recovery_check(token, email):
48    checkout = find_abandoned_checkout(token)
49    if not checkout:
50        return
51    data = {'recovery_url': checkout['abandonedCheckoutUrl'], 'cart_total': checkout['totalPriceV2']['amount'], 'discount': '10%'}
52    send_email(email, TEMPLATE_SECOND, data)
53    logging.info(f'Second recovery email (with discount) sent to {email}')
54
55@app.route('/webhooks/checkout-create', methods=['POST'])
56def checkout_create():
57    hmac_header = request.headers.get('X-Shopify-Hmac-Sha256', '')
58    computed = base64.b64encode(hmac.new(APP_SECRET.encode(), request.get_data(), hashlib.sha256).digest()).decode()
59    if not hmac.compare_digest(computed, hmac_header):
60        abort(401)
61    data = request.get_json()
62    if data.get('email') and data.get('token'):
63        first_recovery_check.apply_async(args=[data['token'], data['email']], countdown=3600)
64    return '', 200
65
66if __name__ == '__main__':
67    app.run(port=3000)