Implementation:Mage ai Mage ai Airtable Source
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, Airtable, Source_Connector |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete tool for extracting data from Airtable tables and bases provided by the Mage integrations source connector framework.
Description
The Airtable source connector extends the base Source class to implement data extraction from Airtable. It implements the discover(), load_data(), and test_connection() methods to connect to an Airtable base, enumerate its tables, infer schemas from sample data, and retrieve all records. The connector uses the pyairtable library through an AirtableConnection wrapper. During discovery, it fetches all records from each table and infers column types (string, array, or object) from the actual data values. Records are flattened so that the Airtable id and createdTime fields are promoted to top-level columns alongside the user-defined fields. Authentication is performed via a personal access token. The connector supports filtering to a single configured table or to a set of selected streams, and defaults to discovering all tables in the base when neither is specified. The replication method is always full-table.
Usage
Use this source connector when building a Mage data pipeline that needs to extract data from Airtable. Configure with a token (personal access token), a base_id (Airtable base identifier), and optionally a table_name to restrict discovery to a single table.
Code Reference
Source Location
- Repository: mage-ai
- File: mage_integrations/mage_integrations/sources/airtable/__init__.py
- Lines: 1-180
Signature
class Airtable(Source):
@property
def base_id(self):
...
@property
def table_name(self):
...
def build_client(self):
...
def test_connection(self) -> None:
...
def load_data(self, stream, **kwargs) -> Generator[List[Dict], None, None]:
...
def discover(self, streams: List[str] = None) -> Catalog:
...
def get_data(self, table: Table) -> List[Dict]:
...
Import
from mage_integrations.sources.airtable import Airtable
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| config | dict | Yes | Configuration dictionary containing token, base_id, and optionally table_name
|
| catalog | Catalog | No | Singer catalog specifying streams to extract |
| state | dict | No | Previous sync state for incremental extraction |
Configuration Parameters
| Name | Type | Required | Description |
|---|---|---|---|
| token | str | Yes | Airtable personal access token for API authentication |
| base_id | str | Yes | Airtable base identifier (e.g., appXXXXXXXXXXXXXX) |
| table_name | str | No | Specific table name to restrict discovery and extraction to a single table |
Outputs
| Name | Type | Description |
|---|---|---|
| catalog | Catalog | Discovered streams with schemas inferred from sample data (from discover()) |
| records | Generator[List[Dict]] | Batches of flattened records with id, createdTime, and field values (from load_data()) |
Usage Examples
from mage_integrations.sources.airtable import Airtable
config = {
"token": "your_airtable_personal_access_token",
"base_id": "appXXXXXXXXXXXXXX",
"table_name": "My Table",
}
source = Airtable(config=config)
# Discover available streams
catalog = source.discover()
# Test connection
source.test_connection()