Error when Upserting Data via API

[jmariano]

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?

[ag314]

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

[jmariano]

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]

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.