Airtable Python Wrapper Documentation

Version:

For more information about the Airtable API see the Airtable API Docs

---

Installation

>>> pip install airtable-python-wrapper

---

Index

Airtable Class

Overview

Airtable Class Instance

>>> airtable = Airtable('base_key', 'table_name')
>>> airtable.get_all()
[{id:'rec123asa23', fields': {'Column': 'Value'}, ...}]

For more information on Api Key and authentication see the Airtable Authentication.

---

Examples

For a full list of available methods see the Airtable class below. For more details on the Parameter filters see the documentation on the available Parameter Filters as well as the Airtable API Docs

Record/Page Iterator:

>>> for page in airtable.get_iter(view='ViewName',sort='COLUMN_A'):
...     for record in page:
...         value = record['fields']['COLUMN_A']

Get all Records:

>>> airtable.get_all(view='ViewName',sort='COLUMN_A')
[{id:'rec123asa23', 'fields': {'COLUMN_A': 'Value', ...}, ... ]

Search:

>>> airtable.search('ColumnA', 'SeachValue')

Formulas:

>>> airtable.get_all(formula="FIND('DUP', {COLUMN_STR})=1")

Insert:

>>> airtable.insert({'First Name', 'John'})

Delete:

>>> airtable.delete('recwPQIfs4wKPyc9D')

You can see the Airtable Class in action in this Jupyter Notebook

---

Return Values

Return Values: when records are returned, they will most often be a list of Airtable records (dictionary) in a format similar to this:

>>> [{
...     "records": [
...         {
...             "id": "recwPQIfs4wKPyc9D",
...             "fields": {
...                 "COLUMN_ID": "1",
...             },
...             "createdTime": "2017-03-14T22:04:31.000Z"
...         },
...         {
...             "id": "rechOLltN9SpPHq5o",
...             "fields": {
...                 "COLUMN_ID": "2",
...             },
...             "createdTime": "2017-03-20T15:21:50.000Z"
...         },
...         {
...             "id": "rec5eR7IzKSAOBHCz",
...             "fields": {
...                 "COLUMN_ID": "3",
...             },
...             "createdTime": "2017-08-05T21:47:52.000Z"
...         }
...     ],
...     "offset": "rec5eR7IzKSAOBHCz"
... }, ... ]

---

Class API

class airtable.Airtable(base_key, table_name, api_key=None)

__init__(base_key, table_name, api_key=None)

If api_key is not provided, AirtableAuth will attempt to use os.environ['AIRTABLE_API_KEY']

batch_delete(record_ids)

Calls delete repetitively, following set API Rate Limit (5/sec) To change the rate limit set value of airtable.API_LIMIT to the time in seconds it should sleep before calling the funciton again.

>>> record_ids = ['recwPQIfs4wKPyc9D', 'recwDxIfs3wDPyc3F']
>>> airtable.batch_delete(records)

Parameters:records (list) – Record Ids to delete Returns:list of records deleted Return type:records (list)

batch_insert(records)

Calls insert repetitively, following set API Rate Limit (5/sec) To change the rate limit use airtable.API_LIMIT = 0.2 (5 per second)

>>> records = [{'Name': 'John'}, {'Name': 'Marc'}]
>>> airtable.batch_insert(records)

Parameters:records (list) – Records to insert Returns:list of added records Return type:records (list)

delete(record_id)

Deletes a record by its id

>>> record = airtable.match('Employee Id', 'DD13332454')
>>> airtable.delete(record['id'])

Parameters:record_id (str) – Airtable record id Returns:Deleted Record Return type:record (dict)

delete_by_field(field_name, field_value, **options)

Deletes first record to match provided field_name and field_value.

>>> record = airtable.delete_by_field('Employee Id', 'DD13332454')

Parameters:

• field_name (str) – Name of field to match (column name).
• field_value (str) – Value of field to match.

Keyword Arguments:  

• view (str, optional) – The name or ID of a view. See ViewParam.
• sort (list, optional) – List of fields to sort by. Default order is ascending. See SortParam.

Returns:

Deleted Record

Return type:

record (dict)

get(record_id)

Retrieves a record by its id

>>> record = airtable.get('recwPQIfs4wKPyc9D')

Parameters:record_id (str) – Airtable record id Returns:Record Return type:record (dict)

get_all(**options)

Retrieves all records repetitively and returns a single list.

>>> airtable.get_all()
>>> airtable.get_all(view='MyView', fields=['ColA', '-ColB'])
>>> airtable.get_all(maxRecords=50)
[{'fields': ... }, ...]

Keyword Arguments:  

• max_records (int, optional) – The maximum total number of records that will be returned. See MaxRecordsParam
• view (str, optional) – The name or ID of a view. See ViewParam.
• fields (str, list, optional) – Name of field or fields to be retrieved. Default is all fields. See FieldsParam.
• sort (list, optional) – List of fields to sort by. Default order is ascending. See SortParam.
• formula (str, optional) – Airtable formula. See FormulaParam.

Returns:

List of Records

Return type:

records (list)

>>> records = get_all(maxRecords=3, view='All')

get_iter(**options)

Record Retriever Iterator

Returns iterator with lists in batches according to pageSize. To get all records at once use get_all

>>> for page in airtable.get_iter():
...     for record in page:
...         print(record)
[{'fields': ... }, ...]

Keyword Arguments:  

• max_records (int, optional) – The maximum total number of records that will be returned. See MaxRecordsParam
• view (str, optional) – The name or ID of a view. See ViewParam.
• page_size (int, optional) – The number of records returned in each request. Must be less than or equal to 100. Default is 100. See PageSizeParam.
• fields (str, list, optional) – Name of field or fields to be retrieved. Default is all fields. See FieldsParam.
• sort (list, optional) – List of fields to sort by. Default order is ascending. See SortParam.
• formula (str, optional) – Airtable formula. See FormulaParam.

Returns:

List of Records, grouped by pageSize

Return type:

iterator (list)

insert(fields)

Inserts a record

>>> record = {'Name': 'John'}
>>> airtable.insert(record)

Parameters:fields (dict) – Fields to insert. Must be dictionary with Column names as Key. Returns:Inserted record Return type:record (dict)

match(field_name, field_value, **options)

Returns first match found in get_all

>>> airtable.match('Name', 'John')
{'fields': {'Name': 'John'} }

Parameters:

• field_name (str) – Name of field to match (column name).
• field_value (str) – Value of field to match.

Keyword Arguments:  

• max_records (int, optional) – The maximum total number of records that will be returned. See MaxRecordsParam
• view (str, optional) – The name or ID of a view. See ViewParam.
• fields (str, list, optional) – Name of field or fields to be retrieved. Default is all fields. See FieldsParam.
• sort (list, optional) – List of fields to sort by. Default order is ascending. See SortParam.

Returns:

First record to match the field_value provided

Return type:

record (dict)

mirror(records, **options)

Deletes all records on table or view and replaces with records.

>>> records = [{'Name': 'John'}, {'Name': 'Marc'}]

>>> record = airtable.,mirror(records)

If view options are provided, only records visible on that view will be deleted.

>>> record = airtable.mirror(records, view='View')
([{'id': 'recwPQIfs4wKPyc9D', ... }], [{'deleted': True, ... }])

Parameters:

records (list) – Records to insert

Keyword Arguments:  

• max_records (int, optional) – The maximum total number of records that will be returned. See MaxRecordsParam
• view (str, optional) – The name or ID of a view. See ViewParam.

Returns:

(new_records, deleted_records)

Return type:

records (tuple)

record_url(record_id)

Builds URL with record id

replace(record_id, fields)

Replaces a record by its record id. All Fields are updated to match the new fields provided. If a field is not included in fields, value will bet set to null. To update only selected fields, use update.

>>> record = airtable.match('Seat Number', '22A')
>>> fields = {'PassangerName': 'Mike', 'Passport': 'YASD232-23'}
>>> airtable.replace(record['id'], fields)

Parameters:

• record_id (str) – Id of Record to update
• fields (dict) – Fields to replace with. Must be dictionary with Column names as Key.

Returns:

New record

Return type:

record (dict)

replace_by_field(field_name, field_value, fields, **options)

Replaces the first record to match field name and value. All Fields are updated to match the new fields provided. If a field is not included in fields, value will bet set to null. To update only selected fields, use update.

Parameters:

• field_name (str) – Name of field to match (column name).
• field_value (str) – Value of field to match.
• fields (dict) – Fields to replace with. Must be dictionary with Column names as Key.

Keyword Arguments:  

• view (str, optional) – The name or ID of a view. See ViewParam.
• sort (list, optional) – List of fields to sort by. Default order is ascending. See SortParam.

Returns:

New record

Return type:

record (dict)

search(field_name, field_value, record=None, **options)

Returns all matching records found in get_all

>>> airtable.search('Gender', 'Male')
[{'fields': {'Name': 'John', 'Gender': 'Male'}, ... ]

Parameters:

• field_name (str) – Name of field to match (column name).
• field_value (str) – Value of field to match.

Keyword Arguments:  

• max_records (int, optional) – The maximum total number of records that will be returned. See MaxRecordsParam
• view (str, optional) – The name or ID of a view. See ViewParam.
• fields (str, list, optional) – Name of field or fields to be retrieved. Default is all fields. See FieldsParam.
• sort (list, optional) – List of fields to sort by. Default order is ascending. See SortParam.

Returns:

All records that matched field_value

Return type:

records (list)

update(record_id, fields)

Updates a record by its record id. Only Fields passed are updated, the rest are left as is.

>>> record = airtable.match('Employee Id', 'DD13332454')
>>> fields = {'Status': 'Fired'}
>>> airtable.update(record['id'], fields)

Parameters:

• record_id (str) – Id of Record to update
• fields (dict) – Fields to update. Must be dictionary with Column names as Key

Returns:

Updated record

Return type:

record (dict)

update_by_field(field_name, field_value, fields, **options)

Updates the first record to match field name and value. Only Fields passed are updated, the rest are left as is.

>>> record = {'Name': 'John', 'Tel': '540-255-5522'}
>>> airtable.update_by_field('Name', 'John', record)

Parameters:

• field_name (str) – Name of field to match (column name).
• field_value (str) – Value of field to match.
• fields (dict) – Fields to update. Must be dictionary with Column names as Key

Keyword Arguments:  

• view (str, optional) – The name or ID of a view. See ViewParam.
• sort (list, optional) – List of fields to sort by. Default order is ascending. See SortParam.

Returns:

Updated record

Return type:

record (dict)

---

Source Code

import sys
import requests
from collections import OrderedDict
from requests.exceptions import HTTPError
import posixpath
import time
from six.moves.urllib.parse import unquote, quote

from .auth import AirtableAuth
from .params import AirtableParams

try:
    IS_IPY = sys.implementation.name == 'ironpython'
except AttributeError:
    IS_IPY = False


class Airtable():

    VERSION = 'v0'
    API_BASE_URL = 'https://api.airtable.com/'
    API_LIMIT = 1.0 / 5  # 5 per second
    API_URL = posixpath.join(API_BASE_URL, VERSION)

    def __init__(self, base_key, table_name, api_key=None):
        """
        If api_key is not provided, :any:`AirtableAuth` will attempt
        to use ``os.environ['AIRTABLE_API_KEY']``
        """
        session = requests.Session()
        session.auth = AirtableAuth(api_key=api_key)
        self.session = session
        self.table_name = table_name
        url_safe_table_name = quote(table_name, safe='')
        self.url_table = posixpath.join(self.API_URL, base_key,
                                        url_safe_table_name)
        self.is_authenticated = self.validate_session(self.url_table)

    def validate_session(self, url):
        response = self.session.get(url, params={'maxRecords': 1})
        if response.ok:
            return True
        elif response.status_code == 404:
            raise ValueError('Invalid base or table name: {}'.format(url))
        else:
            raise ValueError(
                'Authentication failed: {}'.format(response.reason))

    def _process_params(self, params):
        """
        Process params names or values as needed using filters
        """
        new_params = OrderedDict()
        for param_name, param_value in sorted(params.items()):
            param_value = params[param_name]
            ParamClass = AirtableParams._get(param_name)
            new_params.update(ParamClass(param_value).to_param_dict())
        return new_params

    def _process_response(self, response):
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as exc:
            err_msg = str(exc)

            # Reports Decoded 422 Url for better troubleshooting
            # Disabled in IronPython Bug:
            # https://github.com/IronLanguages/ironpython2/issues/242
            if not IS_IPY and response.status_code == 422:
                err_msg = err_msg.replace(response.url, unquote(response.url))
                err_msg += (' (Decoded URL)')

            # Attempt to get Error message from response, Issue #16
            try:
                error_dict = response.json()
            except json.decoder.JSONDecodeError:
                pass
            else:
                if 'error' in error_dict:
                    err_msg += ' [Error: {}]'.format(error_dict['error'])
            raise requests.exceptions.HTTPError(err_msg)
        else:
            return response.json()

    def record_url(self, record_id):
        """ Builds URL with record id """
        return posixpath.join(self.url_table, record_id)

    def _request(self, method, url, params=None, json_data=None):
        response = self.session.request(method, url, params=params,
                                        json=json_data)
        return self._process_response(response)

    def _get(self, url, **params):
        processed_params = self._process_params(params)
        return self._request('get', url, params=processed_params)

    def _post(self, url, json_data):
        return self._request('post', url, json_data=json_data)

    def _put(self, url, json_data):
        return self._request('put', url, json_data=json_data)

    def _patch(self, url, json_data):
        return self._request('patch', url, json_data=json_data)

    def _delete(self, url):
        return self._request('delete', url)

    def get(self, record_id):
        """
        Retrieves a record by its id

        >>> record = airtable.get('recwPQIfs4wKPyc9D')

        Args:
            record_id(``str``): Airtable record id

        Returns:
            record (``dict``): Record
        """
        record_url = self.record_url(record_id)
        return self._get(record_url)

    def get_iter(self, **options):
        """
        Record Retriever Iterator

        Returns iterator with lists in batches according to pageSize.
        To get all records at once use :any:`get_all`

        >>> for page in airtable.get_iter():
        ...     for record in page:
        ...         print(record)
        [{'fields': ... }, ...]

        Keyword Args:
            max_records (``int``, optional): The maximum total number of
                records that will be returned. See :any:`MaxRecordsParam`
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.
            page_size (``int``, optional ): The number of records returned
                in each request. Must be less than or equal to 100.
                Default is 100. See :any:`PageSizeParam`.
            fields (``str``, ``list``, optional): Name of field or fields to
                be retrieved. Default is all fields. See :any:`FieldsParam`.
            sort (``list``, optional): List of fields to sort by.
                Default order is ascending. See :any:`SortParam`.
            formula (``str``, optional): Airtable formula.
                See :any:`FormulaParam`.

        Returns:
            iterator (``list``): List of Records, grouped by pageSize

        """
        offset = None
        while True:
            data = self._get(self.url_table, offset=offset, **options)
            records = data.get('records', [])
            time.sleep(self.API_LIMIT)
            yield records
            offset = data.get('offset')
            if not offset:
                break

    def get_all(self, **options):
        """
        Retrieves all records repetitively and returns a single list.

        >>> airtable.get_all()
        >>> airtable.get_all(view='MyView', fields=['ColA', '-ColB'])
        >>> airtable.get_all(maxRecords=50)
        [{'fields': ... }, ...]

        Keyword Args:
            max_records (``int``, optional): The maximum total number of
                records that will be returned. See :any:`MaxRecordsParam`
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.
            fields (``str``, ``list``, optional): Name of field or fields to
                be retrieved. Default is all fields. See :any:`FieldsParam`.
            sort (``list``, optional): List of fields to sort by.
                Default order is ascending. See :any:`SortParam`.
            formula (``str``, optional): Airtable formula.
                See :any:`FormulaParam`.

        Returns:
            records (``list``): List of Records

        >>> records = get_all(maxRecords=3, view='All')

        """
        all_records = []
        for records in self.get_iter(**options):
            all_records.extend(records)
        return all_records

    def match(self, field_name, field_value, **options):
        """
        Returns first match found in :any:`get_all`

        >>> airtable.match('Name', 'John')
        {'fields': {'Name': 'John'} }

        Args:
            field_name (``str``): Name of field to match (column name).
            field_value (``str``): Value of field to match.

        Keyword Args:
            max_records (``int``, optional): The maximum total number of
                records that will be returned. See :any:`MaxRecordsParam`
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.
            fields (``str``, ``list``, optional): Name of field or fields to
                be retrieved. Default is all fields. See :any:`FieldsParam`.
            sort (``list``, optional): List of fields to sort by.
                Default order is ascending. See :any:`SortParam`.

        Returns:
            record (``dict``): First record to match the field_value provided
        """
        from_name_and_value = AirtableParams.FormulaParam.from_name_and_value
        formula = from_name_and_value(field_name, field_value)
        options['formula'] = formula
        for record in self.get_all(**options):
            return record
        else:
            return {}

    def search(self, field_name, field_value, record=None, **options):
        """
        Returns all matching records found in :any:`get_all`

        >>> airtable.search('Gender', 'Male')
        [{'fields': {'Name': 'John', 'Gender': 'Male'}, ... ]

        Args:
            field_name (``str``): Name of field to match (column name).
            field_value (``str``): Value of field to match.

        Keyword Args:
            max_records (``int``, optional): The maximum total number of
                records that will be returned. See :any:`MaxRecordsParam`
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.
            fields (``str``, ``list``, optional): Name of field or fields to
                be retrieved. Default is all fields. See :any:`FieldsParam`.
            sort (``list``, optional): List of fields to sort by.
                Default order is ascending. See :any:`SortParam`.

        Returns:
            records (``list``): All records that matched ``field_value``

        """
        records = []
        from_name_and_value = AirtableParams.FormulaParam.from_name_and_value
        formula = from_name_and_value(field_name, field_value)
        options['formula'] = formula
        records = self.get_all(**options)
        return records

    def insert(self, fields):
        """
        Inserts a record

        >>> record = {'Name': 'John'}
        >>> airtable.insert(record)

        Args:
            fields(``dict``): Fields to insert.
                Must be dictionary with Column names as Key.

        Returns:
            record (``dict``): Inserted record

        """
        return self._post(self.url_table, json_data={"fields": fields})

    def _batch_request(self, func, iterable):
        """ Internal Function to limit batch calls to API limit """
        responses = []
        for item in iterable:
            responses.append(func(item))
            time.sleep(self.API_LIMIT)
        return responses

    def batch_insert(self, records):
        """
        Calls :any:`insert` repetitively, following set API Rate Limit (5/sec)
        To change the rate limit use ``airtable.API_LIMIT = 0.2``
        (5 per second)

        >>> records = [{'Name': 'John'}, {'Name': 'Marc'}]
        >>> airtable.batch_insert(records)

        Args:
            records(``list``): Records to insert

        Returns:
            records (``list``): list of added records

        """
        return self._batch_request(self.insert, records)

    def update(self, record_id, fields):
        """
        Updates a record by its record id.
        Only Fields passed are updated, the rest are left as is.

        >>> record = airtable.match('Employee Id', 'DD13332454')
        >>> fields = {'Status': 'Fired'}
        >>> airtable.update(record['id'], fields)

        Args:
            record_id(``str``): Id of Record to update
            fields(``dict``): Fields to update.
                Must be dictionary with Column names as Key

        Returns:
            record (``dict``): Updated record
        """
        record_url = self.record_url(record_id)
        return self._patch(record_url, json_data={"fields": fields})

    def update_by_field(self, field_name, field_value, fields, **options):
        """
        Updates the first record to match field name and value.
        Only Fields passed are updated, the rest are left as is.

        >>> record = {'Name': 'John', 'Tel': '540-255-5522'}
        >>> airtable.update_by_field('Name', 'John', record)

        Args:
            field_name (``str``): Name of field to match (column name).
            field_value (``str``): Value of field to match.
            fields(``dict``): Fields to update.
                Must be dictionary with Column names as Key

        Keyword Args:
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.
            sort (``list``, optional): List of fields to sort by.
                Default order is ascending. See :any:`SortParam`.

        Returns:
            record (``dict``): Updated record
        """
        record = self.match(field_name, field_value, **options)
        return {} if not record else self.update(record['id'], fields)

    def replace(self, record_id, fields):
        """
        Replaces a record by its record id.
        All Fields are updated to match the new ``fields`` provided.
        If a field is not included in ``fields``, value will bet set to null.
        To update only selected fields, use :any:`update`.

        >>> record = airtable.match('Seat Number', '22A')
        >>> fields = {'PassangerName': 'Mike', 'Passport': 'YASD232-23'}
        >>> airtable.replace(record['id'], fields)

        Args:
            record_id(``str``): Id of Record to update
            fields(``dict``): Fields to replace with.
                Must be dictionary with Column names as Key.

        Returns:
            record (``dict``): New record
        """
        record_url = self.record_url(record_id)
        return self._put(record_url, json_data={"fields": fields})

    def replace_by_field(self, field_name, field_value, fields, **options):
        """
        Replaces the first record to match field name and value.
        All Fields are updated to match the new ``fields`` provided.
        If a field is not included in ``fields``, value will bet set to null.
        To update only selected fields, use :any:`update`.

        Args:
            field_name (``str``): Name of field to match (column name).
            field_value (``str``): Value of field to match.
            fields(``dict``): Fields to replace with.
                Must be dictionary with Column names as Key.

        Keyword Args:
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.
            sort (``list``, optional): List of fields to sort by.
                Default order is ascending. See :any:`SortParam`.

        Returns:
            record (``dict``): New record
        """
        record = self.match(field_name, field_value, **options)
        return {} if not record else self.replace(record['id'], fields)

    def delete(self, record_id):
        """
        Deletes a record by its id

        >>> record = airtable.match('Employee Id', 'DD13332454')
        >>> airtable.delete(record['id'])

        Args:
            record_id(``str``): Airtable record id

        Returns:
            record (``dict``): Deleted Record
        """
        record_url = self.record_url(record_id)
        return self._delete(record_url)

    def delete_by_field(self, field_name, field_value, **options):
        """
        Deletes first record  to match provided ``field_name`` and
        ``field_value``.

        >>> record = airtable.delete_by_field('Employee Id', 'DD13332454')

        Args:
            field_name (``str``): Name of field to match (column name).
            field_value (``str``): Value of field to match.

        Keyword Args:
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.
            sort (``list``, optional): List of fields to sort by.
                Default order is ascending. See :any:`SortParam`.

        Returns:
            record (``dict``): Deleted Record
        """
        record = self.match(field_name, field_value, **options)
        record_url = self.record_url(record['id'])
        return self._delete(record_url)

    def batch_delete(self, record_ids):
        """
        Calls :any:`delete` repetitively, following set API Rate Limit (5/sec)
        To change the rate limit set value of ``airtable.API_LIMIT`` to
        the time in seconds it should sleep before calling the funciton again.

        >>> record_ids = ['recwPQIfs4wKPyc9D', 'recwDxIfs3wDPyc3F']
        >>> airtable.batch_delete(records)

        Args:
            records(``list``): Record Ids to delete

        Returns:
            records (``list``): list of records deleted

        """
        return self._batch_request(self.delete, record_ids)

    def mirror(self, records, **options):
        """
        Deletes all records on table or view and replaces with records.

        >>> records = [{'Name': 'John'}, {'Name': 'Marc'}]

        >>> record = airtable.,mirror(records)

        If view options are provided, only records visible on that view will
        be deleted.

        >>> record = airtable.mirror(records, view='View')
        ([{'id': 'recwPQIfs4wKPyc9D', ... }], [{'deleted': True, ... }])

        Args:
            records(``list``): Records to insert

        Keyword Args:
            max_records (``int``, optional): The maximum total number of
                records that will be returned. See :any:`MaxRecordsParam`
            view (``str``, optional): The name or ID of a view.
                See :any:`ViewParam`.

        Returns:
            records (``tuple``): (new_records, deleted_records)
        """

        all_record_ids = [r['id'] for r in self.get_all(**options)]
        deleted_records = self.batch_delete(all_record_ids)
        new_records = self.batch_insert(records)
        return (new_records, deleted_records)

    def __repr__(self):
        return '<Airtable table:{}>'.format(self.table_name)

Parameter Filters

Overview

Parameter filters are instantiated internally by using the corresponding keywords.

Filter names (kwargs) can be either the API camelCase name (ie maxRecords) or the snake-case equivalent (max_records).

Refer to the Airtable class to verify which kwargs can be used with each method.

The purpose of these classes is to 1. improve flexibility and ways in which parameter filter values can be passed, and 2. properly format the parameter names and values on the request url.

For more information see the full implementation below.

---

Parameter Filters

class airtable.params.AirtableParams

class FieldsParam(value)

Fields Param

Kwargs:

fields=

Only data for fields whose names are in this list will be included in the records. If you don’t need every field, you can use this parameter to reduce the amount of data transferred.

Usage:

>>> airtable.get(fields='ColumnA')

Multiple Columns:

>>> airtable.get(fields=['ColumnA', 'ColumnB'])

Parameters:fields (str, list) – Name of columns you want to retrieve.

class FormulaParam(value)

Formula Param

Kwargs:

formula= or filterByFormula=

The formula will be evaluated for each record, and if the result is not 0, false, “”, NaN, [], or #Error! the record will be included in the response.

If combined with view, only records in that view which satisfy the formula will be returned. For example, to only include records where COLUMN_A isn’t empty, pass in: "NOT({COLUMN_A}='')"

For more information see Airtable Docs on formulas.

Usage - Text Column is not empty:

>>> airtable.get_all(formula="NOT({COLUMN_A}='')")

Usage - Text Column contains:

>>> airtable.get_all(formula="FIND('SomeSubText', {COLUMN_STR})=1")

Parameters:formula (str) – A valid Airtable formula.

static from_name_and_value(field_name, field_value)

Creates a formula to match cells from from field_name and value

class MaxRecordsParam(value)

Max Records Param

Kwargs:

max_records= or maxRecords=

The maximum total number of records that will be returned.

Usage:

>>> airtable.get_all(max_records=10)

Parameters:max_records (int) – The maximum total number of records that will be returned.

class PageSizeParam(value)

Page Size Param

Kwargs:

page_size= or pageSize=

Limits the maximum number of records returned in each request. Default is 100.

Usage:

>>> airtable.get_all(page_size=50)

Parameters:page_size (int) – The number of records returned in each request. Must be less than or equal to 100. Default is 100.

class SortParam(value)

Sort Param

Kwargs:

sort=

Specifies how the records will be ordered. If you set the view parameter, the returned records in that view will be sorted by these fields.

If sorting by multiple columns, column names can be passed as a list. Sorting Direction is ascending by default, but can be reversed by prefixing the column name with a minus sign -, or passing COLUMN_NAME, DIRECTION tuples. Direction options are asc and desc.

Usage:

>>> airtable.get(sort='ColumnA')

Multiple Columns:

>>> airtable.get(sort=['ColumnA', '-ColumnB'])

Explicit Directions:

>>> airtable.get(sort=[('ColumnA', 'asc'), ('ColumnB', 'desc')])

Parameters:fields (str, list) – Name of columns and directions.

class ViewParam(value)

View Param

Kwargs:

view=

If set, only the records in that view will be returned. The records will be sorted according to the order of the view.

Usage:

>>> airtable.get_all(view='My View')

Parameters:view (str) – The name or ID of a view.

---

Source Code

from collections import OrderedDict


class _BaseParam():

    def __init__(self, value):
        self.value = value

    def to_param_dict(self):
        return {self.param_name: self.value}


class _BaseStringArrayParam(_BaseParam):
    """
    Api Expects Array Of Strings:
    >>> ['FieldOne', 'Field2']

    Requests Params Input:
    >>> params={'fields': ['FieldOne', 'FieldTwo']}

    Requests Url Params Encoding:
    >>> ?fields=FieldOne&fields=FieldTwo

    Expected Url Params:
    >>> ?fields[]=FieldOne&fields[]=FieldTwo
    """

    def to_param_dict(self):
        encoded_param = self.param_name + '[]'
        return {encoded_param: self.value}


class _BaseObjectArrayParam(_BaseParam):
    """
    Api Expects Array of Objects:
    >>> [{field: "UUID", direction: "desc"}, {...}]

    Requests Params Input:
    >>> params={'sort': ['FieldOne', '-FieldTwo']}
    or
    >>> params={'sort': [('FieldOne', 'asc'), ('-FieldTwo', 'desc')]}

    Requests Url Params Encoding:
    >>> ?sort=field&sort=direction&sort=field&sort=direction

    Expected Url Params:
    >>> ?sort[0][field]=FieldOne&sort[0][direction]=asc
    """

    def to_param_dict(self):
        """ Sorts to ensure Order is consistent for Testing """
        param_dict = {}
        for index, dictionary in enumerate(self.value):
            for key, value in dictionary.items():
                param_name = '{param_name}[{index}][{key}]'.format(
                                                    param_name=self.param_name,
                                                    index=index,
                                                    key=key)
                param_dict[param_name] = value
        return OrderedDict(sorted(param_dict.items()))


class AirtableParams():

    class MaxRecordsParam(_BaseParam):
        """
        Max Records Param

        Kwargs:
            ``max_records=`` or ``maxRecords=``

        The maximum total number of records that will be returned.

        Usage:

        >>> airtable.get_all(max_records=10)

        Args:
            max_records (``int``): The maximum total number of records that
                will be returned.


        """

        # Class Input > Output
        # >>> filter = MaxRecordsParam(100)
        # >>> filter.to_param_dict()
        # {'maxRecords: 100}

        param_name = 'maxRecords'
        kwarg = 'max_records'

    class ViewParam(_BaseParam):
        """
        View Param

        Kwargs:
            ``view=``

        If set, only the records in that view will be returned.
        The records will be sorted according to the order of the view.

        Usage:

        >>> airtable.get_all(view='My View')

        Args:
            view (``str``): The name or ID of a view.

        """

        # Class Input > Output
        # >>> filter = ViewParam('Name or Id Of View')
        # >>> filter.to_param_dict()
        # {'view: 'Name or Id Of View'}

        param_name = 'view'
        kwarg = param_name

    class PageSizeParam(_BaseParam):
        """
        Page Size Param

        Kwargs:
            ``page_size=`` or ``pageSize=``

        Limits the maximum number of records returned in each request.
        Default is 100.

        Usage:

        >>> airtable.get_all(page_size=50)

        Args:
            page_size (``int``): The number of records returned in each request.
                Must be less than or equal to 100. Default is 100.

        """
        # Class Input > Output
        # >>> filter = PageSizeParam(50)
        # >>> filter.to_param_dict()
        # {'pageSize: 50}

        param_name = 'pageSize'
        kwarg = 'page_size'

    class FormulaParam(_BaseParam):
        """
        Formula Param

        Kwargs:
            ``formula=`` or ``filterByFormula=``

        The formula will be evaluated for each record, and if the result
        is not 0, false, "", NaN, [], or #Error! the record will be included
        in the response.

        If combined with view, only records in that view which satisfy the
        formula will be returned. For example, to only include records where
        ``COLUMN_A`` isn't empty, pass in: ``"NOT({COLUMN_A}='')"``

        For more information see
        `Airtable Docs on formulas. <https://airtable.com/api>`_

        Usage - Text Column is not empty:

        >>> airtable.get_all(formula="NOT({COLUMN_A}='')")

        Usage - Text Column contains:

        >>> airtable.get_all(formula="FIND('SomeSubText', {COLUMN_STR})=1")

        Args:
            formula (``str``): A valid Airtable formula.

        """

        # Class Input > Output
        # >>> param = FormulaParams("FIND('DUP', {COLUMN_STR})=1")
        # >>> param.to_param_dict()
        # {'formula': "FIND('WW')=1"}

        param_name = 'filterByFormula'
        kwarg = 'formula'

        @staticmethod
        def from_name_and_value(field_name, field_value):
            """
            Creates a formula to match cells from from field_name and value
            """
            if isinstance(field_value, str):
                field_value = "'{}'".format(field_value)

            formula = "{{{name}}}={value}".format(name=field_name,
                                                  value=field_value)
            return formula

    class _OffsetParam(_BaseParam):
        """
        Offset Param

        Kwargs:
            ``offset=``

        If there are more records what was in the response,
        the response body will contain an offset value.
        To fetch the next page of records,
        include offset in the next request's parameters.

        This is used internally by :any:`get_all` and :any:`get_iter`.

        Usage:

        >>> airtable.get_iter(offset='recjAle5lryYOpMKk')

        Args:
            record_id (``str``, ``list``):

        """
        # Class Input > Output
        # >>> filter = _OffsetParam('recqgqThAnETLuH58')
        # >>> filter.to_param_dict()
        # {'offset: 'recqgqThAnETLuH58'}

        param_name = 'offset'
        kwarg = param_name

    class FieldsParam(_BaseStringArrayParam):
        """
        Fields Param

        Kwargs:
            ``fields=``

        Only data for fields whose names are in this list will be included in
        the records. If you don't need every field, you can use this parameter
        to reduce the amount of data transferred.

        Usage:

        >>> airtable.get(fields='ColumnA')

        Multiple Columns:

        >>> airtable.get(fields=['ColumnA', 'ColumnB'])

        Args:
            fields (``str``, ``list``): Name of columns you want to retrieve.

        """

        # Class Input > Output
        # >>> param = FieldsParam(['FieldOne', 'FieldTwo'])
        # >>> param.to_param_dict()
        # {'fields[]': ['FieldOne', 'FieldTwo']}

        param_name = 'fields'
        kwarg = param_name

    class SortParam(_BaseObjectArrayParam):
        """
        Sort Param

        Kwargs:
            ``sort=``

        Specifies how the records will be ordered. If you set the view
        parameter, the returned records in that view will be sorted by these
        fields.

        If sorting by multiple columns, column names can be passed as a list.
        Sorting Direction is ascending by default, but can be reversed by
        prefixing the column name with a minus sign ``-``, or passing
        ``COLUMN_NAME, DIRECTION`` tuples. Direction options
        are ``asc`` and ``desc``.

        Usage:

        >>> airtable.get(sort='ColumnA')

        Multiple Columns:

        >>> airtable.get(sort=['ColumnA', '-ColumnB'])

        Explicit Directions:

        >>> airtable.get(sort=[('ColumnA', 'asc'), ('ColumnB', 'desc')])

        Args:
            fields (``str``, ``list``): Name of columns and directions.

        """

        # Class Input > Output
        # >>> filter = SortParam([{'field': 'col', 'direction': 'asc'}])
        # >>> filter.to_param_dict()
        # {'sort[0]['field']: 'col', sort[0]['direction']: 'asc'}

        param_name = 'sort'
        kwarg = param_name

        def __init__(self, value):
            # Wraps string into list to avoid string iteration
            if hasattr(value, 'startswith'):
                value = [value]

            self.value = []
            direction = 'asc'

            for item in value:
                if not hasattr(item, 'startswith'):
                    field_name, direction = item
                else:
                    if item.startswith('-'):
                        direction = 'desc'
                        field_name = item[1:]
                    else:
                        field_name = item

                sort_param = {'field': field_name, 'direction': direction}
                self.value.append(sort_param)

    @classmethod
    def _discover_params(cls):
        """
        Returns a dict where filter keyword is key, and class is value.
        To handle param alias (maxRecords or max_records), both versions are
        added.
        """

        try:
            return cls.filters
        except AttributeError:
            filters = {}
            for param_class_name in dir(cls):
                param_class = getattr(cls, param_class_name)
                if hasattr(param_class, 'kwarg'):
                    filters[param_class.kwarg] = param_class
                    filters[param_class.param_name] = param_class
            cls.filters = filters
        return cls.filters

    @classmethod
    def _get(cls, kwarg_name):
        """ Returns a Param Class Instance, by its kwarg or param name """
        param_classes = cls._discover_params()
        try:
            param_class = param_classes[kwarg_name]
        except KeyError:
            raise ValueError('invalid param keyword {}'.format(kwarg_name))
        else:
            return param_class

Airtable Authentication

Overview

Authentication is handled by the Airtable class. The class can handle authentication automatically if the environment variable AIRTABLE_API_KEY is set with your api key.

>>> airtable = Airtable(base_key, table_name)

Alternatively, you can pass the key explicitly:

>>> airtable = Airtable(base_key, table_name, api_key='yourapikey')

Note

You can also use this class to handle authentication for you if you are making your own wrapper:

>>> auth = AirtableAuth(api_key)
>>> response = requests.get('https://api.airtable.com/v0/{basekey}/{table_name}', auth=auth)

---

Authentication Class

class airtable.auth.AirtableAuth(api_key=None)

__init__(api_key=None)

Authentication used by Airtable Class

Parameters:api_key (str) – Airtable API Key. Optional. If not set, it will look for enviroment variable AIRTABLE_API_KEY

---

Source Code

from __future__ import absolute_import
import os
import requests


class AirtableAuth(requests.auth.AuthBase):

    def __init__(self, api_key=None):
        """
        Authentication used by Airtable Class

        Args:
            api_key (``str``): Airtable API Key. Optional.
                If not set, it will look for
                enviroment variable ``AIRTABLE_API_KEY``
        """
        try:
            self.api_key = api_key or os.environ['AIRTABLE_API_KEY']
        except KeyError:
            raise KeyError('Api Key not found. Pass api_key as a kwarg \
                            or set an env var AIRTABLE_API_KEY with your key')

    def __call__(self, request):
        auth_token = {'Authorization': 'Bearer {}'.format(self.api_key)}
        request.headers.update(auth_token)
        return request

• Index
• Module Index

Release Notes

Release Notes

Questions

Post them over in the project’s Github Page

---

Contribute

https://www.github.com/gtalarico/airtable-python-wrapper

License

MIT License