Genesis Foods GraphQL API Playground
- Doug Bell
Owned by Doug Bell
Genesis Foods GraphQL API Playground Repository
GitHub - esha/genesis-foods-api-examples: Genesis Foods GraphQL API Playground
Python
Configuration Example (config.ini)
[api]
endpoint = [https://api-preview.trustwell.com/genesis | https://api.trustwell.com/genesis]
api_key = [YOUR_API_KEY]
[files]
output_file = graphql_responses.json
input_file = input_data.jsonImport / Export
Bulk Export
Exports are limited to 10,000 entries at any one time. To further refine an export set the searchText value in the input variables.
import requests
import configparser
import json
import os
# Read the configuration file
config = configparser.ConfigParser()
config.read('config.ini')
# Retrieve the API settings
endpoint = config.get('api', 'endpoint')
api_key = config.get('api', 'api_key')
# Retrieve the output file path from the configuration
output_file = config.get('files', 'output_file')
# Ensure the file is written to the local directory
file_path = os.path.join(os.getcwd(), output_file)
# Delete the file if it already exists
if os.path.exists(file_path):
os.remove(file_path)
print(f"Existing file '{file_path}' has been deleted.")
# Define the headers
headers = {
"X-API-KEY": api_key,
"Content-Type": "application/json"
}
# Define the GraphQL query
query = """
query($input: FoodSearchInput!){
foods{
search(input: $input) {
foodSearchResults{
id
name
modified
created
versionName
eshaCode
foodType
product
supplier
versionHistoryId
}
totalCount
pageInfo{
cursor
hasNextPage
startCursor
endCursor
}
}
}
}
"""
def run_query(graphql_query, variables):
"""Run a GraphQL query against the Genesis API with variables."""
response = requests.post(
endpoint,
json={'query': graphql_query, 'variables': variables},
headers=headers
)
if response.status_code == 200:
return response.json()
else:
print(f"Request failed with status code {response.status_code}")
print(response.text)
return None
def export(graphql_query, food_type):
first_entry = True
"""Iterate through each character in the corpus and update the searchText."""
with open(file_path, "a") as file:
file.write("[\n") # Write the opening bracket for the JSON array
variables = {
"input": {
"searchText": '*',
"foodTypes": [food_type],
"itemSourceFilter": "Customer",
"versionFilter": "All",
"first": 10000,
"after": 0
}
}
print(f"Running query...")
result = run_query(graphql_query, variables)
if result:
total_count = result.get("data", {}).get("foods", {}).get("search", {}).get("totalCount", 0)
print(f"Found {total_count} results")
if total_count > 0:
if not first_entry:
file.write(",\n") # Add a comma before each entry - except the first
json.dump(result, file, indent=4) # Append the JSON response to the file
first_entry = False
else:
print(f"No results found. Skipping write.")
file.write("\n]") # Write the closing bracket for the JSON array
if __name__ == "__main__":
export(query, 'Ingredient') # Ingredient or Recipe
print(f"Complete. Exported results to {file_path}")
Bulk Import
import requests
import configparser
import json
import os
import uuid
# Read the configuration file
config = configparser.ConfigParser()
config.read('config.ini')
# Retrieve the API settings
endpoint = config.get('api', 'endpoint')
api_key = config.get('api', 'api_key')
# Define the headers with API Key
headers = {
"X-API-KEY": api_key,
"Content-Type": "application/json"
}
# Define the GraphQL mutation for creating an ingredient
mutation = """
mutation($input : CreateFoodInput!){
foods{
create(input:$input)
{
food{
id
name
definingAmount{
quantity{
value
}
unit{
id
name
}
}
conversions{
from{
quantity{
value
}
unit{
id
name
}
}
to{
quantity{
value
}
unit{
id
name
}
}
}
}
}
}
}
"""
def run_mutation(food, food_type):
"""Run the GraphQL mutation to create an ingredient."""
input_data = {
"input": {
"name": food.get("name") + ' ' + food_type + ':' + str(uuid.uuid4()),
"foodType": food_type,
# "definingAmount": food.get("definingAmount"),
# "conversions": food.get("conversions")
}
}
response = requests.post(endpoint, json={"query": mutation, "variables": input_data}, headers=headers)
if response.status_code == 200:
return response.json()
else:
print(f"Error: {response.status_code} - {response.text}")
return None
def bulk_import(food_type, file_path):
"""Load JSON data from a file and run the mutation for each ingredient."""
with open(file_path, 'r') as f:
data = json.load(f)
for food in data:
print(f"Importing {food_type}: {food.get('name')}")
result = run_mutation(food, food_type)
if result and "errors" in result:
print(f"Error while creating {food_type}: {food.get('name')}")
print(result["errors"])
elif result:
created_food = result.get("data", {}).get("foods", {}).get("create", {}).get("food", {})
print(f"Created {food_type}: {created_food.get('name')} (ID: {created_food.get('id')})")
def main():
# Load the input file path from config
input_file_path = config.get('files', 'input_file')
# Ensure the file exists
if not os.path.exists(input_file_path):
print(f"Input file '{input_file_path}' does not exist.")
return
# Import ingredients
bulk_import('Ingredient', input_file_path)
# Import recipies
bulk_import('Recipe', input_file_path)
if __name__ == "__main__":
main()