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=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 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=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]