Webhooks¶

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

pyAirtable allows you to create and manage webhooks and to retrieve webhook payloads using a straightforward API within the Base class.

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='2023-07-01T00:00:00.000Z'
)

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="2023-07-01T00:00:00.000Z",
        specification: WebhookSpecification(...)
    )
]

Return type: List[Webhook]

Base.webhook(webhook_id)[source]

Build a single webhook or raises KeyError 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

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="2022-02-01T21:25:05.663Z",
    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]

class pyairtable.models.WebhookNotification[source]

Represents the value that Airtable will POST to the webhook’s notification URL.

This will not contain the full webhook payload; it will only contain the IDs of the base and the webhook which triggered the notification. You will need to use Webhook.payloads to retrieve the actual payloads describing the change(s) which triggered the webhook.

You will also need some way to persist the cursor of the webhook payload, so that on subsequent calls you do not retrieve the same payloads again.

Usage:

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_db.get(f"cursor_{event.webhook}", 0)) + 1
    for payload in webhook.payloads(cursor=cursor):
        # ...do stuff...
        your_db.set(f"cursor_{event.webhook}", payload.cursor)
    return ("", 204)  # intentionally empty response

See Webhook notification delivery for more information on how these payloads are structured.

classmethod 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.

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.