如何向Airtable单元格追加图片URL？解决批量接收覆盖问题

当用户一次性发送多张图片时，WATI会发起多个独立POST请求到你的接口，这些请求几乎同时到达服务器。每个请求都会先读取Airtable中该用户的旧图片URL，追加新URL后再写回。但因为Airtable更新存在延迟，后续请求读取到的仍是最初的旧值，最终只有最后一个请求的URL被保留——这是典型的竞态条件问题。

直接通过Airtable API的公式更新语法，让Airtable自身完成追加操作，从根源避免竞态。无需先读取旧值，直接在更新指令中完成拼接。

修改代码示例 #

首先实现原子追加的更新方法：

import requests

AIRTABLE_API_KEY = '你的API密钥'
BASE_ID = '你的Base ID'
TABLE_NAME = '你的表名'

def update_image_url_append(record_id, new_url):
    url = f'https://api.airtable.com/v0/{BASE_ID}/{TABLE_NAME}/{record_id}'
    headers = {
        'Authorization': f'Bearer {AIRTABLE_API_KEY}',
        'Content-Type': 'application/json'
    }
    # 用Airtable公式实现：为空则设新值，否则追加换行+新URL
    data = {
        "fields": {
            "image_fetched": f"IF({{image_fetched}}='', '{new_url}', {{image_fetched}} & '\\n' & '{new_url}')"
        }
    }
    response = requests.patch(url, json=data, headers=headers)
    response.raise_for_status()

然后修改接口逻辑：

@app.route('/kc', methods=['POST', 'GET'])
def execute():
    data = request.json
    senderID = data['waId']
    name = data['senderName']

    if data['type'] == 'image':
        new_image_url = data.get('data')
        print("URL received ", new_image_url)

        # 仅获取记录ID，无需读取旧URL
        record_id, _ = at.get_field(senderID, "image_fetched")
        print(record_id)

        # 调用原子追加方法
        update_image_url_append(record_id, new_image_url)

    return data

针对同一用户ID加互斥锁，确保同一用户的请求串行处理，避免并发读取旧值。用线程锁结合字典实现用户专属锁：

from threading import Lock

# 全局字典存储每个用户的锁，键为senderID
user_locks = {}
# 保护字典的全局锁
global_lock = Lock()

@app.route('/kc', methods=['POST', 'GET'])
def execute():
    data = request.json
    senderID = data['waId']
    name = data['senderName']

    if data['type'] == 'image':
        new_image_url = data.get('data')
        print("URL received ", new_image_url)

        # 获取当前用户的专属锁
        with global_lock:
            if senderID not in user_locks:
                user_locks[senderID] = Lock()
        user_lock = user_locks[senderID]

        # 同一用户的请求串行执行，确保读取最新值
        with user_lock:
            id, existing_image_url = at.get_field(senderID, "image_fetched")
            print(id, existing_image_url)
            if existing_image_url is None:
                    existing_image_url = ""

            image_urls = existing_image_url.split("\n")
            # 可选：避免重复添加同一URL
            if new_image_url not in image_urls:
                image_urls.append(new_image_url)
            image_array = "\n".join(image_urls)

            print(image_array)
            at.update_image_url(id, image_array)

    return data

部分WhatsApp API会将多张图片的URL打包在一个请求的数组中，而非分多次请求。可以打印data字段确认，如果是数组则直接批量追加：

if data['type'] == 'image':
    new_image_data = data.get('data')
    if isinstance(new_image_data, list):
        for url in new_image_data:
            print("URL received ", url)
            update_image_url_append(record_id, url)
    else:
        new_image_url = new_image_data
        print("URL received ", new_image_url)
        update_image_url_append(record_id, new_image_url)

内容的提问来源于stack exchange，提问作者Dummy Cron