Created on September 12, 2024
This script demonstrates a simple integration between Shopify and Airtable, automatically syncing new orders from your Shopify store to an Airtable base. Each order is stored as a single row in Airtable, including order details, customer information, and a summary of products purchased.
Get your Shopify Access Token with the following API access scopes
read_orders
read_customers
read_products
Get your Airtable Personal Access Token with the following token permissions:
data.records:write
schema.bases:read
appXXXXXXXXXXXXXX
Test Table
becomes Test%20Table
)https://your-store-name.myshopify.com
)In your Airtable base, create a table named "Orders" with the following fields:
In your Shopify admin panel:
import os
import requests
from datetime import datetime
SHOPIFY_ACCESS_TOKEN = os.environ.get('SHOPIFY_ACCESS_TOKEN')
SHOPIFY_SHOP_NAME = os.environ.get('SHOPIFY_SHOP_NAME')
SHOPIFY_API_VERSION = '2024-07'
AIRTABLE_PERSONAL_ACCESS_TOKEN = os.environ.get('AIRTABLE_PERSONAL_ACCESS_TOKEN')
AIRTABLE_BASE_ID = os.environ.get('AIRTABLE_BASE_ID')
AIRTABLE_TABLE_NAME = os.environ.get('AIRTABLE_TABLE_NAME', 'Orders')
def handler(request):
payload = request['body']
order_id = payload.get('id')
shopify_url = f"https://{SHOPIFY_SHOP_NAME}.myshopify.com/admin/api/{SHOPIFY_API_VERSION}/orders/{order_id}.json"
headers = {
"X-Shopify-Access-Token": SHOPIFY_ACCESS_TOKEN,
"Content-Type": "application/json"
}
shopify_response = requests.get(shopify_url, headers=headers)
if not shopify_response.ok:
raise requests.exceptions.RequestException(f"Shopify error: {shopify_response.status_code}, {shopify_response.text}")
order = shopify_response.json()['order']
product_count = len(order['line_items'])
product_info = '|'.join(f"{item.get('sku', 'N/A')}:{item['quantity']}" for item in order['line_items'])
airtable_data = {
"fields": {
"Order Number": str(order['order_number']),
"Order Date": datetime.fromisoformat(order['created_at'].replace('Z', '+00:00')).strftime('%Y-%m-%d'),
"Total Price": float(order['total_price']),
"Customer Name": f"{order['customer']['first_name']} {order['customer']['last_name']}",
"Customer Email": order['customer']['email'],
"Number of Products": product_count,
"Product SKUs and Quantities": product_info
}
}
airtable_url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{AIRTABLE_TABLE_NAME}"
headers = {
"Authorization": f"Bearer {AIRTABLE_PERSONAL_ACCESS_TOKEN}",
"Content-Type": "application/json"
}
airtable_response = requests.post(airtable_url, json=airtable_data, headers=headers)
if not airtable_response.ok:
raise requests.exceptions.RequestException(f"Airtable error: {airtable_response.status_code}, {airtable_response.text}")
print("Processed Successfully")