Airtable Sync

Category: Operations

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

Mirror an Airtable table into Postgres — every page of it, upserted so re-running is safe.

The two things that make Airtable different

1. It paginates, and quietly. Airtable returns at most 100 records per request plus an offset token, and keeps returning one until you reach the end. Fetch once and you silently get the first 100 rows and no indication there were more — the classic way an Airtable sync looks fine in testing and loses data in production.

This pack uses the native pagination block, which follows the token to the last page and flattens every page into one array:

url: ".../{base}/{table}?pageSize=100&offset=${page}"
pagination:
  payload_key: records      # where the rows live
  next_cursor_key: offset   # where the next cursor lives
  max_pages: 100            # safety bound

${page} marks where the cursor goes. On the first request that whole parameter is dropped — Airtable has no cursor yet and rejects a made-up one — and pagination stops when Airtable omits offset, which is how it signals the last page.

Requires an Air Pipe engine with cursor-pagination first-page handling (1.37.2+).

2. Records are nested. An Airtable record is { id, fields: {...} }, not flat, so columns read as a|body::fields.Name|. A field the record doesn't set is simply absent, which is why every column here carries a ->default(...).

Endpoints

Method	Route	Purpose
POST	`/airtable/sync`	Pull every page and upsert each record.
GET	`/airtable/records`	List synced records, highest score first (`?limit=`).
POST	`/api/seed`	Create the `airtable_records` table (from `seed.yml`).

Setup

Create a personal access token at Airtable → Developer hub, with data.records:read on your base.
Find your base id (appXXXXXXXXXXXXXX) — it's in the API docs URL for that base.
Set the managed variables:
- DATABASE_URL, AIRTABLE_TOKEN, AIRTABLE_BASE, AIRTABLE_TABLE
Seed the table: POST /api/seed.
Sync: POST /airtable/sync → { "synced": 3 }.

Mapping your own columns

The upsert names Airtable columns directly. For a table with Name / Email / Score:

a|body::fields.Name->default('')->single_quote|
a|body::fields.Score->default(0)|

Rename those to match your table, keeping the ->default(...) on each — an Airtable field that has never been set on a record is absent rather than null, and the default is what stops the insert failing on it.

Keeping it in sync

POST /airtable/sync is idempotent — it upserts on the Airtable record id, so run it as often as you like. To run it automatically, add a schedule: block (see the Scheduled Cleanup pack):

schedule:
  cron: "0 * * * *"
  enabled: true

Deletions in Airtable are not propagated: a record removed there stays in Postgres. If you need that, compare the synced ids against the table and delete the difference.

Configuration

config.yml

name: AirtableSync

docs: true

# Sync an Airtable table into Postgres, then push edits back.
#
# Two Airtable-specific things this handles:
#   1. PAGINATION — Airtable returns 100 records at a time plus an `offset` token,
#      and you must keep calling until the offset stops coming back. The native
#      `pagination` block follows it automatically and flattens every page into
#      one array, so partial syncs aren't silently truncated at 100 rows.
#   2. SHAPE — records are `{ id, fields: {...} }`, not flat. Column values live
#      under `fields`, and a field the record doesn't set is simply absent.
#
# Required managed variables:
#   DATABASE_URL     — Postgres connection string
#   AIRTABLE_TOKEN   — personal access token (Airtable → Developer hub)
#   AIRTABLE_BASE    — base id, looks like appXXXXXXXXXXXXXX
#   AIRTABLE_TABLE   — table name or id

global:
  variables:
    airtable_api: "https://api.airtable.com/v0"

  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # POST /airtable/sync
  # Pulls EVERY page and upserts each record.
  airtable/sync:
    output: http
    method: POST
    summary: Pull every Airtable record (all pages) and upsert into Postgres
    description: >
      Follows Airtable's offset pagination to the last page, then upserts each
      record on its Airtable id so re-running is safe.
    tags: [airtable, sync, pagination, postgres]
    response_example: |-
      { "synced": 3 }

    actions:
      # `pagination` follows the offset token and flattens every page into one
      # array — without it you silently get only the first 100 records.
      # `${page}` marks where the cursor goes. On the FIRST request the whole
      # `offset=${page}` parameter is dropped (Airtable has no cursor yet and 422s
      # on a made-up one); afterwards it is replaced with the `offset` Airtable
      # returned. Pagination stops when Airtable omits `offset` — the last page.
      - name: Fetch
        http:
          url: "a|var::airtable_api|/a|ap_var::AIRTABLE_BASE|/a|ap_var::AIRTABLE_TABLE|?pageSize=100&offset=${page}"
          method: GET
          bearer_auth: a|ap_var::AIRTABLE_TOKEN|
          pagination:
            payload_key: records      # where the rows live in each page
            next_cursor_key: offset   # where the next cursor lives
            max_pages: 100            # safety bound: 100 pages x 100 records

      # One upsert per record. Inside a lookup the current record is `body`, so
      # the Airtable shape reads as a|body::fields.|<Column>|.
      - name: UpsertEach
        run_when_succeeded: [Fetch]
        lookup: a|Fetch::body.records|
        lookup_concurrency: 5
        actions:
          - name: Upsert
            database: main
            query: |-
              INSERT INTO airtable_records (airtable_id, name, email, score, synced_at)
              VALUES (
                a|body::id->single_quote|,
                a|body::fields.Name->default('')->single_quote|,
                a|body::fields.Email->default('')->single_quote|,
                a|body::fields.Score->default(0)|,
                NOW()
              )
              ON CONFLICT (airtable_id) DO UPDATE SET
                name      = EXCLUDED.name,
                email     = EXCLUDED.email,
                score     = EXCLUDED.score,
                synced_at = NOW()
              RETURNING airtable_id

      - name: Count
        run_when_succeeded: [UpsertEach]
        database: main
        query: "SELECT COUNT(*) AS synced FROM airtable_records"
        post_transforms:
          - extract_value: "[0]"

      - name: Respond
        run_when_succeeded: [Count]
        json_output: |-
          { "synced": a|Count::synced| }

  # GET /airtable/records?limit=25
  airtable/records:
    output: http
    method: GET
    summary: List synced records
    tags: [airtable, postgres]

    actions:
      - name: ReadParams
        input: a|params|
        hide_data_on_success: true

      - name: ListRecords
        run_when_succeeded: [ReadParams]
        database: main
        query: |-
          SELECT airtable_id, name, email, score, synced_at
            FROM airtable_records
           ORDER BY score DESC
           LIMIT a|ReadParams::limit->default(25)|

seed.yml

name: AirtableSyncSeed
description: Creates the airtable_records table (idempotent). Safe to re-run — resets to empty.

docs: true

# Required managed variables:
#   DATABASE_URL — Postgres connection string

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # POST /api/seed
  api/seed:
    output: http
    method: POST

    actions:
      - name: CreateTable
        database: main
        hide_data_on_success: true
        query: |
          CREATE TABLE IF NOT EXISTS airtable_records (
            airtable_id  TEXT         PRIMARY KEY,
            name         TEXT         NOT NULL DEFAULT '',
            email        TEXT         NOT NULL DEFAULT '',
            score        INTEGER      NOT NULL DEFAULT 0,
            synced_at    TIMESTAMPTZ  NOT NULL DEFAULT NOW()
          );

      - name: Reset
        database: main
        run_when_succeeded: [CreateTable]
        hide_data_on_success: true
        query: "TRUNCATE TABLE airtable_records;"

      - name: Respond
        run_when_succeeded: [Reset]
        json_output: |-
          { "status": "seeded" }

schema.sql

-- Records mirrored from an Airtable table, keyed on the Airtable record id so a
-- re-sync updates in place instead of duplicating.
CREATE TABLE IF NOT EXISTS airtable_records (
  airtable_id  TEXT         PRIMARY KEY,
  name         TEXT         NOT NULL DEFAULT '',
  email        TEXT         NOT NULL DEFAULT '',
  score        INTEGER      NOT NULL DEFAULT 0,
  synced_at    TIMESTAMPTZ  NOT NULL DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS idx_airtable_records_score ON airtable_records (score DESC);

Airtable Sync

The two things that make Airtable different​

Endpoints​

Setup​

Mapping your own columns​

Keeping it in sync​

Configuration​

config.yml​

seed.yml​

schema.sql​