Help

Error when Upserting Data via API

1580 3
cancel
Showing results for 
Search instead for 
Did you mean: 
jmariano
4 - Data Explorer
4 - Data Explorer

I am working on an integration that pulls data from a source, matches it to an account in AirTable, based on a common field - "Domain" - and then appends the Account record with data from the other source.

I have gotten this to work on small tables with only 4-5 Accounts, however when I updated my script and pointed it at our production data, I am running into the following error.

"Error: 422 - {"error":{"type":"INVALID_VALUE_FOR_COLUMN","message":"Cannot update more than one record for fields to merge on"}}"

Do I need to modify the script to only update one record at a time?

3 Replies 3
ag314
6 - Interface Innovator
6 - Interface Innovator

Probably. If you share your script (after redacting any keys or webhook urls) it will be easier to figure out the issue.

import requests
import json
import base64
import datetime

AFFINITY_API_KEY = '**********'
LIST_ID = '*******'
AIRTABLE_API_KEY = '**********'
AIRTABLE_BASE_ID = '**************'
AIRTABLE_TABLE_NAME = 'Accounts'

AFFINITY_API_BASE_URL = 'https://api.affinity.co'
AIRTABLE_API_BASE_URL = f'https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{AIRTABLE_TABLE_NAME}'

# Affinity step: Fetch organization data using organization ID
def get_affinity_organization(org_id😞
url = f'{AFFINITY_API_BASE_URL}/organizations/{org_id}'
 
# Use empty username and API key as password for Basic Auth
api_key_encoded = base64.b64encode(f':{AFFINITY_API_KEY}'.encode('utf-8')).decode('utf-8')
 
headers = {
'Authorization': f'Basic {api_key_encoded}',
'Content-Type': 'application/json'
}
 
params = {'with_interaction_dates': True, 'with_interaction_fields': True} # Include interaction fields
 
response = requests.get(url, headers=headers, params=params)
 
if response.status_code == 200:
return response.json()
else:
print(f"Error: {response.status_code} - {response.text}")
return None

# Fetch organization IDs from the specific list
list_id = '*******'
org_ids = get_organization_ids_from_list(list_id)

# Affinity step: Fetch organization data using organization ID
interaction_data = []
for org_id in org_ids:
organization = get_affinity_organization(org_id)
domain = organization.get('domain')
last_email_date = organization['interaction_dates'].get('last_email_date') if 'interaction_dates' in organization else None
first_email_date = organization['interaction_dates'].get('first_email_date') if 'interaction_dates' in organization else None
last_event_date = organization['interaction_dates'].get('last_event_date') if 'interaction_dates' in organization else None
next_event_date = organization['interaction_dates'].get('next_event_date') if 'interaction_dates' in organization else None

interaction_data.append({
'org_id': org_id,
'domain': domain,
'last_email_date': last_email_date,
'first_email_date': first_email_date,
'last_event_date': last_event_date,
'next_event_date': next_event_date
})

# Affinity step: Fetch organization data using organization ID
interaction_data = []
for org_id in org_ids:
organization = get_affinity_organization(org_id)
domain = organization.get('domain')
last_email_date = organization['interaction_dates'].get('last_email_date') if 'interaction_dates' in organization else None
first_email_date = organization['interaction_dates'].get('first_email_date') if 'interaction_dates' in organization else None
last_event_date = organization['interaction_dates'].get('last_event_date') if 'interaction_dates' in organization else None
next_event_date = organization['interaction_dates'].get('next_event_date') if 'interaction_dates' in organization else None

interaction_data.append({
'org_id': org_id,
'domain': domain,
'last_email_date': last_email_date,
'first_email_date': first_email_date,
'last_event_date': last_event_date,
'next_event_date': next_event_date
})

# Airtable step: Update record in Airtable
def update_airtable_records(records😞
url = f'{AIRTABLE_API_BASE_URL}'

headers = {
'Authorization': f'Bearer {AIRTABLE_API_KEY}',
'Content-Type': 'application/json'
}

data = {
"records": records,
"typecast": True,
"performUpsert": {
"fieldsToMergeOn": ["Domain"]
}
}

response = requests.patch(url, headers=headers, json=data)

if response.status_code == 200:
return response.json()
else:
print(f"Error: {response.status_code} - {response.text}")
return None

# Prepare records for updating in Airtable
airtable_records = []
for org_data in interaction_data:
domain = org_data.get('domain')
last_email_date = org_data.get('last_email_date')
first_email_date = org_data.get('first_email_date') # Additional field 1 from Affinity
last_event_date = org_data.get('last_event_date') # Additional field 2 from Affinity
next_event_date = org_data.get('next_event_date') # Additional field 3 from Affinity

if domain and last_email_date:
airtable_record = {
"fields": {
"Domain": domain,
"Last Email - AFFINITY": last_email_date,
"First Email - AFFINITY": first_email_date, # Additional field 1 in Airtable
"Last Meeting - AFFINITY": last_event_date, # Additional field 2 in Airtable
"Next Meeting - AFFINITY": next_event_date, # Additional field 3 in Airtable
}
}
airtable_records.append(airtable_record)

# Update records in Airtable
updated_records = update_airtable_records(airtable_records)

if updated_records:
created_records = updated_records.get("createdRecords", [])
updated_record_ids = updated_records.get("updatedRecords", [])

for record_id in created_records:
print(f"Created Airtable record")
ag314
6 - Interface Innovator
6 - Interface Innovator

Using the Airtable API to update multiple records at once, using Upserts, will fail if there are multiple matches found for the merge field, in your case, Domain. If you try without the upsert and include the id’s to update, do you still get the same failure? Also, note that there is a limit of 10 records that you can update per call. A quick look at your code and I’m not seeing any throttling of the data being written.