Is anyone having issues performing constrained searches against their Bubble API using Python?
I have followed the bubble documentation to fetch my data with great success. Now, however, I would like to filter my results using constraints as specified here. Below is the code I use, which, if I omit the constraints, returns all of my data, but if I include them, it fails every time.
import requests
import urllib.parse
import pandas as pd
session = requests.Session()
#API KEY as defined in the settings tab, so that the script has full access to the data
API_KEY = 'SECRET_KEY'
#base endpoint, see https://bubble.io/reference#API.get_api.Endpoint
base_url = 'https://myapp.bubbleapps.io/version-test/api/1.1/obj/line_item'
#Query initial parameters. We do not send a limit parameter (default is 100)
#we keep making calls till we have no remaining items to fetch
def get_data(constraints = False):
remaining = 1
cursor = 0
dfs = []
while remaining:
#data we send with the search. Search constraints would be here
if constraints:
params = {'cursor': cursor, 'api_token': API_KEY, 'constraints': constraints}
else:
params = {'cursor': cursor, 'api_token': API_KEY}
url = base_url + '?' + urllib.parse.urlencode(params)
response = session.get(url)
print(response.url)
if response.status_code != 200:
print('Error with status code {}'.format(response.status_code))
return "You're a sad boy"
chunk = response.json()['response']
remaining = chunk['remaining'] # Set remaining to check if should repeat next loop
dfs.append(pd.DataFrame(chunk['results']))
cursor += chunk['count']
return pd.concat(dfs)
search_options = [{'key': 'quantity',
'constraint_type': 'greater than',
'value': 2}]
df = get_data(search_options)
I’ve tried many variations of constraints including capitalizations, and against different field types and tables without success. My database / api permissions seam correct (afterall, I do get all my data returned already) So I’m wondering if the URL encoding via urllib is to blame. I know that the original code example for this was written in Python 2 (as obviated by the print statements) so I updated it to use Python3s urllib.parse.urlencode as recommended here). I even tried encoding using the requests library but that didn’t work either. I really need to constrain my searches.
Also, I’m really surprised to see that Bubble Devs didn’t specify / recommend to put time delays between Data API requests… using something like time.sleep(0.1). Since not having delays can cause all sorts of problems for your servers? May I suggest that update to your your documentation? Also, I lovingly encourage an update to Python3 examples since Python2 is deprecated as of January 1st 2020. You’re welcome to use my code example above (if we can ever get the constraints to work).
Turns out the reason this wasn’t working was because my URL encoding was wrong. Using a tip to json.dump the constraint parameters, I’ve got this working. The Python3x code is provided below for your enjoyment:
import pandas as pd
import json
import time
session = requests.Session()
#API KEY as defined in the settings tab, so that the script has full access to the data
API_KEY = 'YOUR_SECRET_KEY'
#base endpoint, see https://bubble.io/reference#API.get_api.Endpoint
base_url = 'https://YOU_APP.io/version-test/api/1.1/obj/YOUR_DATA_THING'
#Query initial parameters. We do not send a limit parameter (default is 100)
#we keep making calls till we have no remaining items to fetch
def get_data(constraints = False):
remaining = 1
cursor = 0
dfs = []
while remaining:
#data we send with the search. Search constraints would be here
headers = {"api_token": API_KEY, "cursor": str(cursor)}
if constraints:
payload = {"constraints": json.dumps(constraints)}
response = session.get(base_url, headers = headers, params=payload)
else:
response = session.get(base_url, headers = headers)
print(response.url)
if response.status_code != 200:
print('Error with status code {}'.format(response.status_code))
return "You're a sad boy"
chunk = response.json()['response']
remaining = chunk['remaining'] # Set remaining to check if should repeat next loop
dfs.append(pd.DataFrame(chunk['results']))
cursor += chunk['count']
time.sleep(0.1)
return pd.concat(dfs)
# search_options = [{"key": "quantity","constraint_type": "greater than","value": 2}]
search_options = [{'key': 'Created Date',
'constraint_type': 'greater than',
'value': '2020-01-26T04:32:14.906Z'}]
df = get_data(search_options)
Posting my Python question in this topic because it’s similar and @zelus_pudding might have some insight.
But you also changed the way you passed your API token. I’ve tried it both ways and can’t get it to work in a POST request. If I pass api_token in the headers, I get “401 Unauthorized”. If I pass it in the payload, I get “unrecognized key: api_token”
base_url = 'https://myapp.bubbleapps.io/version-test/api/1.1/obj/user'
def get_users():
# For GET, passing the api token in the params works:
data = dict(limit=50,
cursor=0,
api_token=bubble_api_token)
resp = requests.get(base_url, data)
def add_user(name, email):
# For POST, neither works:
data = dict(Name=name,
email=email)
# api_token=bubble_api_token)
headers = {"api_token": bubble_api_token}
resp = requests.post(base_url, data, headers=headers)
This should get you started - it’s a snippet of what I use in production but tweaked to erase my secret bits. You’ll have to set your own URL, API Key, and update the search_bubble_by_constraints call to search on your thing rather than investor as I have it below
import json
import pandas as pd
import requests
import time
BUBBLE_API_KEY = "ENTER_API_KEY_HERE"
bubble_url = "bubble_api_endpoint"
class bubble_response_error(Exception):
def __init__(self, message):
self.message = message
def generate_error_message(response):
error_payload = {
"target_url": response.url,
"response_status_code": response.status_code,
"response_json": response.json()}
error_message = response=json.dumps(error_payload)
return error_message
def search_bubble_by_constraints(session, bubble_url, bubble_thing, constraints=False, backoff_seconds=0.01):
# Query initial parameters. We do not send a limit parameter (default is 100)
# we keep making calls till we have no remaining items to fetch
url = bubble_url + '/' + bubble_thing
remaining = 1
cursor = 0
dfs = []
while remaining:
if constraints:
payload = {"cursor": str(cursor),
"api_token": BUBBLE_API_KEY,
"constraints": json.dumps(constraints)}
response = session.get(url, params=payload)
else:
payload = {"cursor": str(cursor),
"api_token": BUBBLE_API_KEY}
response = session.get(url, params=payload)
if response.status_code != 200:
raise bubble_response_error(generate_error_message(response))
chunk = response.json()['response']
# Set remaining to check if should repeat next loop
remaining = chunk['remaining']
dfs.append(pd.DataFrame(chunk['results']))
cursor += chunk['count']
time.sleep(backoff_seconds)
return pd.concat(dfs)
with requests.Session() as session:
df_investors = search_bubble_by_constraints(session, bubble_url, 'investor', False)
The Post request is similar. Below, the row of a pandas dataframe with columns named after the same columns in my bubble db are converted into json before being attached as data. Then I enrich that data payload with Agent and Territories key/value pairs.
