Python client for the Airtable API providing comprehensive database operations, ORM functionality, enterprise features, and testing utilities
—
Fundamental Airtable API access through the Api, Base, and Table classes. These provide the foundation for all pyAirtable operations including authentication, connection management, and basic CRUD operations.
The Api class manages authentication, HTTP connections, retry strategies, and provides factory methods for accessing bases and tables.
class Api:
def __init__(self, api_key: str, *,
timeout: Optional[tuple[int, int]] = None,
retry_strategy: Optional[Union[bool, object]] = True,
endpoint_url: str = "https://api.airtable.com",
use_field_ids: bool = False):
"""
Initialize API connection.
Parameters:
- api_key: Airtable API key or personal access token
- timeout: (connect_timeout, read_timeout) in seconds
- retry_strategy: Retry configuration (True for default, False to disable)
- endpoint_url: API endpoint URL (for debugging/proxy)
- use_field_ids: Use field IDs instead of names in responses
"""
def whoami(self) -> dict:
"""
Get current user ID and scopes.
Returns:
dict with 'id' and optional 'scopes' keys
"""
def base(self, base_id: str, *, validate: bool = False, force: bool = False) -> Base:
"""
Get Base instance.
Parameters:
- base_id: Airtable base ID (starts with 'app')
- validate: Validate base exists via API call
- force: Force refresh of cached metadata
Returns:
Base instance
"""
def table(self, base_id: str, table_name: str, *,
validate: bool = False, force: bool = False) -> Table:
"""
Get Table instance directly.
Parameters:
- base_id: Airtable base ID (starts with 'app')
- table_name: Table name or ID (starts with 'tbl')
- validate: Validate table exists via API call
- force: Force refresh of cached metadata
Returns:
Table instance
"""
def bases(self, *, force: bool = False) -> list[Base]:
"""
List all accessible bases.
Parameters:
- force: Force refresh of cached base list
Returns:
List of Base instances
"""
def create_base(self, workspace_id: str, name: str, tables: list[dict]) -> Base:
"""
Create a new base.
Parameters:
- workspace_id: Workspace ID where base will be created
- name: Base name
- tables: List of table schema dictionaries
Returns:
Base instance for the created base
"""
def workspace(self, workspace_id: str) -> Workspace:
"""
Get Workspace instance.
Parameters:
- workspace_id: Workspace ID (starts with 'wsp')
Returns:
Workspace instance
"""
def enterprise(self, enterprise_account_id: str) -> Enterprise:
"""
Get Enterprise instance (Enterprise plans only).
Parameters:
- enterprise_account_id: Enterprise account ID
Returns:
Enterprise instance
"""
@property
def api_key(self) -> str:
"""Current API key or access token."""
# HTTP methods
def request(self, method: str, url: str, **kwargs) -> dict:
"""Make authenticated API request."""
def get(self, url: str, **kwargs) -> dict:
"""Make GET request."""
def post(self, url: str, **kwargs) -> dict:
"""Make POST request."""
def patch(self, url: str, **kwargs) -> dict:
"""Make PATCH request."""
def delete(self, url: str, **kwargs) -> dict:
"""Make DELETE request."""Base class represents an Airtable base and provides access to tables, metadata, webhooks, and collaboration features.
class Base:
def __init__(self, api: Api, base_id: str, *,
name: Optional[str] = None,
permission_level: Optional[str] = None):
"""
Initialize Base instance.
Parameters:
- api: Api instance
- base_id: Base ID (starts with 'app')
- name: Base name (if known)
- permission_level: User's permission level on base
"""
@property
def id(self) -> str:
"""Base ID."""
@property
def name(self) -> Optional[str]:
"""Base name (if available)."""
def table(self, id_or_name: str, *,
validate: bool = False, force: bool = False) -> Table:
"""
Get Table instance.
Parameters:
- id_or_name: Table name or ID
- validate: Validate table exists
- force: Force refresh metadata
Returns:
Table instance
"""
def tables(self, *, force: bool = False) -> list[Table]:
"""
Get all tables in base.
Parameters:
- force: Force refresh base schema
Returns:
List of Table instances
"""
def create_table(self, name: str, fields: list[dict],
description: Optional[str] = None) -> Table:
"""
Create new table in base.
Parameters:
- name: Table name (must be unique)
- fields: List of field schema dictionaries
- description: Table description (max 20k characters)
Returns:
Table instance for created table
"""
def schema(self, *, force: bool = False) -> object:
"""
Get base schema with all tables and fields.
Parameters:
- force: Force refresh cached schema
Returns:
BaseSchema object with tables and field information
"""
def webhooks(self) -> list[object]:
"""
Get all webhooks configured for this base.
Returns:
List of Webhook objects
"""
def webhook(self, webhook_id: str) -> object:
"""
Get specific webhook by ID.
Parameters:
- webhook_id: Webhook ID
Returns:
Webhook object
"""
def add_webhook(self, notify_url: str, spec: Union[dict, object]) -> object:
"""
Create webhook for base.
Parameters:
- notify_url: URL to receive webhook notifications
- spec: Webhook specification dict or WebhookSpecification object
Returns:
CreateWebhookResponse with webhook ID and secret
"""
def delete(self) -> None:
"""Delete base (Enterprise only)."""Table class provides direct access to records and table-level operations including CRUD operations, schema management, and field creation.
class Table:
def __init__(self, api_key: Optional[str], base: Union[Base, str],
table_name: Union[str, object]):
"""
Initialize Table instance.
Parameters:
- api_key: API key (deprecated, use Api.table() instead)
- base: Base instance or base ID
- table_name: Table name, ID, or TableSchema object
"""
@property
def id(self) -> str:
"""Table ID (fetched from API if needed)."""
@property
def name(self) -> str:
"""Table name."""
@property
def base(self) -> Base:
"""Base instance containing this table."""
@property
def api(self) -> Api:
"""API instance used by this table."""
def schema(self, *, force: bool = False) -> object:
"""
Get table schema with field definitions.
Parameters:
- force: Force refresh cached schema
Returns:
TableSchema object with field information
"""
def create_field(self, name: str, field_type: str,
description: Optional[str] = None,
options: Optional[dict] = None) -> object:
"""
Create new field in table.
Parameters:
- name: Field name (must be unique in table)
- field_type: Airtable field type (e.g., 'singleLineText', 'number')
- description: Field description
- options: Field-specific options dict
Returns:
FieldSchema object for created field
"""from pyairtable import Api
# Initialize with API key
api = Api('your_personal_access_token')
# Check authentication
user_info = api.whoami()
print(f"Connected as user: {user_info['id']}")
# List available bases
bases = api.bases()
for base in bases:
print(f"Base: {base.name} ({base.id})")# Get base instance
base = api.base('app1234567890abcde')
# Get base schema
schema = base.schema()
print(f"Base has {len(schema.tables)} tables")
# List all tables
tables = base.tables()
for table in tables:
print(f"Table: {table.name} ({table.id})")
# Get specific table
table = base.table('My Table Name')
# or by table ID
table = base.table('tbl1234567890abcde')# Get table schema
schema = table.schema()
print(f"Table has {len(schema.fields)} fields")
# Create new field
new_field = table.create_field(
name='Priority',
field_type='singleSelect',
options={
'choices': [
{'name': 'High', 'color': 'redBright'},
{'name': 'Medium', 'color': 'yellowBright'},
{'name': 'Low', 'color': 'greenBright'}
]
}
)
print(f"Created field: {new_field.name}")from pyairtable import Api
from pyairtable.api.retrying import retry_strategy
# Custom timeout and retry configuration
api = Api(
'your_token',
timeout=(5, 30), # 5s connect, 30s read timeout
retry_strategy=retry_strategy(
total=5, # max retries
backoff_factor=1.0 # exponential backoff
)
)
# Disable retries
api = Api('your_token', retry_strategy=False)
# Use field IDs instead of names
api = Api('your_token', use_field_ids=True)Install with Tessl CLI
npx tessl i tessl/pypi-pyairtable