pyAirtable Documentation

When modeling bidirectional links (e.g., Company has Employees, and Employee has a Company), you may encounter circular dependencies. To resolve this, you can define the target model in the field using one of three methods:

1. Fully qualified module path: Provide a string like "your.module.Model".
2. Class name string: Provide just the class name string (e.g., "Person") if it's in the same module.
3. Sentinel value: Use F.LinkSelf to point to the model where the field is defined.

LinkField arguments: LinkField(attribute_name, airtable_field_name, model_definition).

from pyairtable.orm import Model, fields as F

class Company(Model):
    class Meta: ...
    name = F.TextField("Name")
    employees = F.LinkField("Employees", "path.to.Person")  # Option 1: Full path

class Person(Model):
    class Meta: ...
    name = F.TextField("Name")
    company = F.SingleLinkField[Company]("Company", Company)
    manager = F.SingleLinkField["Person"]("Manager", "Person")  # Option 2: Class name
    reports = F.LinkField["Person"]("Reports", F.LinkSelf)      # Option 3: LinkSelf

To avoid making hundreds of API calls when traversing nested linked records, you can use memoization. This pre-fetches and reuses model instances.

Global/Model Level

Set memoize = True in the Meta configuration of a model to enable it by default.

Per-call Level

Pass memoize=True to retrieval methods to pre-fetch records into the cache.

Memoization Support Table

# Pre-fetching all books and authors to avoid N+1 queries
books = Book.all(memoize=True)
authors = Author.all(memoize=True)

for author in authors:
    for book in author.books:
        print(book.title) # Uses memoized instance

Airtable's Webhooks API allows you to receive programmatic notifications when Airtable data or metadata changes. The integration workflow with pyAirtable follows these steps:

1. Create: Use Base.add_webhook to register a new webhook with a specific URL.
2. Receive: Airtable sends POST notifications to your provided URL.
3. Validate: Use WebhookNotification.from_request to verify the authenticity of the incoming notification using the request body, the X-Airtable-Content-MAC header, and your webhook secret.
4. Process: Use Webhook.payloads to retrieve the actual data changes (payloads).

Important: Cursor Management To avoid processing the same data multiple times or missing data during interruptions, you must persist the cursor of the webhook payload. When calling webhook.payloads(cursor=...), you should store the latest payload.cursor + 1 in a database so your next execution knows where to start.

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

# Example workflow for receiving and processing webhooks
@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"]
    
    # 1. Validate the notification
    event = WebhookNotification.from_request(body, header, secret)
    
    airtable = Api(app.config["AIRTABLE_API_KEY"])
    webhook = airtable.base(event.base.id).webhook(event.webhook.id)
    
    # 2. Retrieve the last saved cursor from your database
    cursor = int(your_database.get(event.webhook, 0)) + 1

    # 3. Iterate through new payloads using the cursor
    for payload in webhook.payloads(cursor=cursor):
        process_payload(payload)
        # 4. Persist the next cursor to prevent duplicate processing
        your_database.set(event.webhook, payload.cursor + 1)

    return ("", 204)

The pyairtable.Api class manages the connection to Airtable, and the pyairtable.Table class provides methods to interact with specific tables.

Common operations include:

• api.table(base_id, table_id): Connects to a specific table.
• table.all(): Retrieves all records.
• table.create(fields): Creates a new record.
• table.update(record_id, fields): Updates an existing record.
• table.delete(record_id): Deletes a record.

import os
from pyairtable import Api

# Initialize the API connection using an environment variable
api = Api(os.environ['AIRTABLE_API_KEY'])

# Access a specific table
table = api.table('appExampleBaseId', 'tblExampleTableId')

# Retrieve all records
records = table.all()
print(records)

# Create a new record
new_record = table.create({"Name": "Bob"})
print(new_record)

# Update an existing record
table.update("recwAcQdqwe21asdf", {"Name": "Robert"})

# Delete a record
table.delete("recwAcQdqwe21asdf")