Skip to content

Instantly share code, notes, and snippets.

@BFTrick
Last active October 18, 2024 21:56
Show Gist options
  • Save BFTrick/89b07fda34c218e9d989c0ba2dd7db24 to your computer and use it in GitHub Desktop.
Save BFTrick/89b07fda34c218e9d989c0ba2dd7db24 to your computer and use it in GitHub Desktop.
Fetch SKU and available inventory from Ship Hero
#!/usr/bin/env python3
from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
from gql.transport.exceptions import TransportQueryError
import json
import time
import os
bearer = ADDYOURTOKENHERE
query_template = """query($cursor: String) {
products {
request_id
complexity
data(first: 500, after: $cursor) {
pageInfo {
endCursor
hasNextPage
}
edges {
node {
id
sku
active
warehouse_products {
warehouse_id
available
}
}
}
}
}
}"""
_transport = RequestsHTTPTransport(
url='https://public-api.shiphero.com/graphql',
use_json=True,
)
_transport.headers = {
"User-Agent": "Mozilla/5.0 (X11; buntu; Linux x86_64; rv:58.0) Gecko/0100101 Firefox/58.0",
"Authorization": "Bearer {}".format(bearer),
"content-type": "application/json",
}
client = Client(transport=_transport, fetch_schema_from_transport=True)
query = gql(query_template)
batch_number = 1
cursor = None
while True:
try:
print(f"Executing query for batch {batch_number}")
variables = {"cursor": cursor}
result = client.execute(query, variable_values=variables)
filename = f'output-{batch_number}.txt'
print(f"Writing to {filename}")
with open(filename, 'w') as f:
f.write("SKU,Available\n")
for product in result['products']['data']['edges']:
if product['node']['active']:
for warehouse_product in product['node']['warehouse_products']:
if warehouse_product['warehouse_id'] == "V2FyZWhvdXNlOjc4OTMx":
f.write("{},{}\n".format(product['node']['sku'], warehouse_product['available']))
break
f.write(json.dumps(result, indent=2))
print(f"Successfully wrote to {filename}")
batch_number += 1
# Update cursor for the next batch
cursor = result['products']['data']['pageInfo']['endCursor']
if not result['products']['data']['pageInfo']['hasNextPage']:
print("No more products to fetch.")
break
except TransportQueryError as e:
error_data = e.errors[0]
print(f"Error occurred: {error_data}")
if error_data['code'] == 30:
print(f"Rate limit hit. Waiting for {error_data['time_remaining']} seconds.")
time.sleep(120)
else:
raise e
except Exception as e:
print(f"An unexpected error occurred: {e}")
break
@BFTrick
Copy link
Author

BFTrick commented Oct 18, 2024

You can run this script from the terminal with python3 ship-hero-inventory.py

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment