tessl/pypi-pyairtable

Python client for the Airtable API providing comprehensive database operations, ORM functionality, enterprise features, and testing utilities

—

Pending

Overview

Eval results

Files

Record Operations

Name: tessl/pypi-pyairtable
Author: tessl

Advanced record handling capabilities including batch operations, upserts, pagination, filtering, and sorting. These operations provide efficient data management for single records and bulk operations.

Capabilities

Single Record Operations

Basic CRUD operations for individual records with full control over field updates and retrieval options.

def get(self, record_id: str, **options) -> dict:
    """
    Retrieve single record by ID.
    
    Parameters:
    - record_id: Record ID (starts with 'rec')
    - cell_format: 'json' (default) or 'string'
    - time_zone: Timezone for datetime fields (e.g., 'America/New_York')
    - user_locale: Locale for formatting (e.g., 'en-US')
    - use_field_ids: Return field IDs instead of names
    
    Returns:
    Record dict with 'id', 'createdTime', and 'fields' keys
    """

def create(self, fields: dict, typecast: bool = False, 
           use_field_ids: Optional[bool] = None) -> dict:
    """
    Create single record.
    
    Parameters:
    - fields: Field name/ID -> value mapping
    - typecast: Automatically convert field values to correct types
    - use_field_ids: Return field IDs in response
    
    Returns:
    Created record dict
    """

def update(self, record_id: str, fields: dict, 
           replace: bool = False, typecast: bool = False,
           use_field_ids: Optional[bool] = None) -> dict:
    """
    Update single record.
    
    Parameters:
    - record_id: Record ID to update
    - fields: Field updates (name/ID -> value mapping)
    - replace: If True, replace all fields (PUT). If False, merge (PATCH)
    - typecast: Automatically convert field values
    - use_field_ids: Return field IDs in response
    
    Returns:
    Updated record dict
    """

def delete(self, record_id: str) -> dict:
    """
    Delete single record.
    
    Parameters:
    - record_id: Record ID to delete
    
    Returns:
    Dict with 'id' and 'deleted': True
    """

Batch Record Operations

Efficient bulk operations for creating, updating, and deleting multiple records in single API calls.

def batch_create(self, records: list[dict], typecast: bool = False,
                 use_field_ids: Optional[bool] = None) -> list[dict]:
    """
    Create multiple records in batches.
    
    Parameters:
    - records: List of field dicts to create
    - typecast: Automatically convert field values
    - use_field_ids: Return field IDs in response
    
    Returns:
    List of created record dicts
    """

def batch_update(self, records: list[dict], replace: bool = False,
                 typecast: bool = False, 
                 use_field_ids: Optional[bool] = None) -> list[dict]:
    """
    Update multiple records in batches.
    
    Parameters:
    - records: List of dicts with 'id' and 'fields' keys
    - replace: If True, replace all fields (PUT). If False, merge (PATCH)
    - typecast: Automatically convert field values
    - use_field_ids: Return field IDs in response
    
    Returns:
    List of updated record dicts
    """

def batch_delete(self, record_ids: list[str]) -> list[dict]:
    """
    Delete multiple records in batches.
    
    Parameters:
    - record_ids: List of record IDs to delete
    
    Returns:
    List of dicts with 'id' and 'deleted': True
    """

def batch_upsert(self, records: list[dict], key_fields: list[str],
                 replace: bool = False, typecast: bool = False,
                 use_field_ids: Optional[bool] = None) -> dict:
    """
    Create or update records based on key fields.
    
    Parameters:
    - records: List of record dicts (with optional 'id' field)
    - key_fields: Field names to match existing records
    - replace: If True, replace all fields. If False, merge
    - typecast: Automatically convert field values
    - use_field_ids: Return field IDs in response
    
    Returns:
    Dict with 'createdRecords', 'updatedRecords', and 'records' lists
    """

Record Retrieval and Pagination

Flexible record retrieval with filtering, sorting, pagination, and field selection options.

def all(self, **options) -> list[dict]:
    """
    Retrieve all matching records.
    
    Parameters:
    - view: View name or ID to filter records
    - page_size: Records per page (1-100, default 100)
    - max_records: Maximum total records to return
    - fields: List of field names/IDs to include
    - sort: List of sort specifications
    - formula: Formula string to filter records
    - cell_format: 'json' (default) or 'string'
    - time_zone: Timezone for datetime fields
    - user_locale: Locale for formatting
    - use_field_ids: Return field IDs instead of names
    
    Returns:
    List of all matching record dicts
    """

def iterate(self, **options) -> Iterator[list[dict]]:
    """
    Iterate through records in pages.
    
    Parameters: Same as all()
    
    Yields:
    List of records for each page
    """

def first(self, **options) -> Optional[dict]:
    """
    Get first matching record.
    
    Parameters: Same as all() (automatically sets max_records=1)
    
    Returns:
    First matching record dict or None
    """

Usage Examples

Single Record Operations

from pyairtable import Api

api = Api('your_token')
table = api.table('base_id', 'table_name')

# Create record
record = table.create({
    'Name': 'John Doe',
    'Email': 'john@example.com',
    'Age': 30
})
print(f"Created: {record['id']}")

# Get record
retrieved = table.get(record['id'])
print(f"Name: {retrieved['fields']['Name']}")

# Update record (partial update)
updated = table.update(record['id'], {
    'Age': 31,
    'Status': 'Active'
})

# Replace record (full replacement)
replaced = table.update(record['id'], {
    'Name': 'Jane Doe',
    'Email': 'jane@example.com'
}, replace=True)

# Delete record
deleted = table.delete(record['id'])
print(f"Deleted: {deleted['deleted']}")

Batch Operations

# Batch create multiple records
records_to_create = [
    {'Name': 'Alice', 'Department': 'Engineering'},
    {'Name': 'Bob', 'Department': 'Sales'},
    {'Name': 'Carol', 'Department': 'Marketing'}
]

created_records = table.batch_create(records_to_create)
print(f"Created {len(created_records)} records")

# Batch update
updates = [
    {'id': created_records[0]['id'], 'fields': {'Status': 'Active'}},
    {'id': created_records[1]['id'], 'fields': {'Status': 'Pending'}}
]

updated_records = table.batch_update(updates)

# Batch upsert (create or update based on Name field)
upsert_data = [
    {'Name': 'Alice', 'Department': 'Senior Engineering'},  # Updates existing
    {'Name': 'David', 'Department': 'HR'}  # Creates new
]

result = table.batch_upsert(upsert_data, key_fields=['Name'])
print(f"Created: {len(result['createdRecords'])}, Updated: {len(result['updatedRecords'])}")

# Batch delete
record_ids = [r['id'] for r in created_records]
deleted = table.batch_delete(record_ids)
print(f"Deleted {len(deleted)} records")

Advanced Retrieval

# Get all records with filtering and sorting
records = table.all(
    view='Active Users',
    fields=['Name', 'Email', 'Department'],
    sort=[
        {'field': 'Department', 'direction': 'asc'},
        {'field': 'Name', 'direction': 'asc'}
    ],
    max_records=50
)

# Use formula for complex filtering
from pyairtable.formulas import match, AND, OR

# Records where Department is 'Engineering' and Status is 'Active'
engineering_active = table.all(
    formula=match({
        'Department': 'Engineering',
        'Status': 'Active'
    })
)

# More complex formula
complex_filter = OR(
    AND(match({'Department': 'Engineering'}), match({'Level': ('>=', 3)})),
    match({'Department': 'Management'})
)
records = table.all(formula=complex_filter)

# Paginate through large datasets
for page in table.iterate(page_size=50):
    print(f"Processing {len(page)} records")
    for record in page:
        process_record(record)

# Get first matching record
first_user = table.first(
    formula=match({'Status': 'New'}),
    sort=[{'field': 'Created', 'direction': 'asc'}]
)
if first_user:
    print(f"Oldest new user: {first_user['fields']['Name']}")

Type Casting and Field Options

# Enable automatic type conversion
record = table.create({
    'Name': 'Test User',
    'Age': '25',  # String will be converted to number
    'Active': 'true',  # String will be converted to boolean
    'Join Date': '2023-01-15'  # String will be converted to date
}, typecast=True)

# Use field IDs instead of names
api_with_field_ids = Api('your_token', use_field_ids=True)
table_with_ids = api_with_field_ids.table('base_id', 'table_name')

record = table_with_ids.create({
    'fld1234567890abcd': 'Value using field ID'
})

# Custom formatting options
record = table.get('record_id', 
    cell_format='string',  # Get formatted strings instead of raw values
    time_zone='America/New_York',  # Convert dates to specific timezone
    user_locale='en-US'  # Use specific locale for formatting
)

Error Handling for Batch Operations

import requests

try:
    # Batch operations can partially succeed
    result = table.batch_create(large_dataset)
except requests.exceptions.HTTPError as e:
    # Check if it's a rate limit or validation error
    if e.response.status_code == 429:
        print("Rate limited, wait and retry")
    elif e.response.status_code == 422:
        print("Validation error in data")
        error_details = e.response.json()
        print(f"Error: {error_details}")

Install with Tessl CLI