Webhooks¶

Airtable’s Webhooks API offers a configurable and flexible way to receive programmatic notifications of changes to Airtable data and metadata.

The basic flow for webhooks is:

Use Base.add_webhook to create a webhook.
Airtable will POST notifications to the webhook URL you provided.
Use WebhookNotification.from_request to validate each notification.
Use Webhook.payloads to retrieve new payloads after the notification.

This means it is technically possible to ignore webhook notifications altogether and to simply poll a webhook periodically for new payloads. However, this increases the likelihood of running into Airtable’s API rate limits.

When using webhooks, you need some way to persist the cursor of the webhook payload, so that you do not retrieve the same payloads again on subsequent calls, even if your job is interrupted in the middle of processing a list of payloads.

For example:

from flask import Flask, request
from pyairtable import Api
from pyairtable.models import WebhookNotification

app = Flask(__name__)

@app.route("/airtable-webhook", methods=["POST"])
def airtable_webhook():
    body = request.data
    header = request.headers["X-Airtable-Content-MAC"]
    secret = app.config["AIRTABLE_WEBHOOK_SECRET"]
    event = WebhookNotification.from_request(body, header, secret)
    airtable = Api(app.config["AIRTABLE_API_KEY"])
    webhook = airtable.base(event.base.id).webhook(event.webhook.id)
    cursor = int(your_database.get(event.webhook, 0)) + 1

    for payload in webhook.payloads(cursor=cursor):
        process_payload(payload)  # probably enqueue a background job
        your_database.set(event.webhook, payload.cursor + 1)

    return ("", 204)  # intentionally empty response

Methods¶

The following methods will be most commonly used for working with payloads. You can read the full documentation at pyairtable.models.webhook.

Base.add_webhook(notify_url, spec)[source]

Create a webhook on the base with the given webhooks specification.

The return value will contain a unique secret that must be saved in order to validate payloads as they are sent to your notification endpoint. If you do not save this, you will have no way of confirming that payloads you receive did, in fact, come from Airtable.

For more on how to validate notifications to your webhook, see WebhookNotification.from_request().

Usage:

>>> base.add_webhook(
...     "https://example.com",
...     {
...         "options": {
...             "filters": {
...                 "dataTypes": ["tableData"],
...             }
...         }
...     }
... )
CreateWebhookResponse(
    id='ach00000000000001',
    mac_secret_base64='c3VwZXIgZHVwZXIgc2VjcmV0',
    expiration_time=datetime.datetime(...)
)

Raises

pydantic.ValidationError – If the dict provided is invalid.

Parameters

notify_url (str) – The URL where Airtable will POST all event notifications.
spec (Union[WebhookSpecification, Dict[Any, Any]]) – The configuration for the webhook. It is easiest to pass a dict that conforms to the webhooks specification but you can also provide WebhookSpecification.

Return type

CreateWebhookResponse

Base.webhooks()[source]

Retrieve all the base’s webhooks (see: List webhooks).

Usage:

>>> base.webhooks()
[
    Webhook(
        id='ach00000000000001',
        are_notifications_enabled=True,
        cursor_for_next_payload=1,
        is_hook_enabled=True,
        last_successful_notification_time=None,
        notification_url="https://example.com",
        last_notification_result=None,
        expiration_time=datetime.datetime(...),
        specification: WebhookSpecification(...)
    )
]

Return type: List[Webhook]

Base.webhook(webhook_id)[source]

Build a single webhook or raises MissingRecordError if the given ID is invalid.

Airtable’s API does not permit retrieving a single webhook, so this function will call webhooks() and simply return one item from the list.

Return type: Webhook

classmethod WebhookNotification.from_request(body, header, secret)[source]

Validate a request body and X-Airtable-Content-MAC header using the secret returned when the webhook was created.

Parameters

body (str) – The full request body sent over the wire.
header (str) – The request’s X-Airtable-Content-MAC header.
secret (Union[bytes, str]) – The MAC secret provided when the webhook was created. If str, it’s assumed this is still base64-encoded; if bytes, it’s assumed that this has been decoded.

Returns

An instance parsed from the provided request body.

Return type

WebhookNotification

Raises

ValueError – if the header and body do not match the secret.

Webhook.payloads(cursor=1, *, limit=None)[source]

Iterate through all payloads on or after the given cursor. See WebhookPayload. Each payload will contain an extra attribute, cursor, which you will need to store if you want to later resume retrieving payloads after that point.

For more details on the mechanisms of retrieving webhook payloads, or to find more information about the data structures you’ll get back, see List webhook payloads.

Parameters

cursor (int, default: 1) – The cursor of the first webhook payload to retrieve.
limit (Optional[int], default: None) – The number of payloads to yield before stopping. If not provided, will retrieve all remaining payloads.

Usage:

>>> webhook = Base.webhook("ach00000000000001")
>>> iter_payloads = webhook.payloads()
>>> next(iter_payloads)
WebhookPayload(
    timestamp=datetime.datetime(...),
    base_transaction_number=4,
    payload_format="v0",
    action_metadata=ActionMetadata(
        source="client",
        source_metadata={
            "user": {
                "id": "usr00000000000000",
                "email": "foo@bar.com",
                "permissionLevel": "create"
            }
        }
    ),
    changed_tables_by_id={},
    created_tables_by_id={},
    destroyed_table_ids=["tbl20000000000000", "tbl20000000000001"],
    error=None,
    error_code=None,
    cursor=1
)

Return type: Iterator[WebhookPayload]