AirTable Interface

speechline.utils.airtable.AirTable

AirTable table interface.

Parameters:

Name	Type	Description	Default
url	str	URL of AirTable table.	required

Exceptions:

Type	Description
OSError	AIRTABLE_API_KEY environment is not set.

Source code in speechline/utils/airtable.py

class AirTable:
    """
    AirTable table interface.

    Args:
        url (str):
            URL of AirTable table.

    Raises:
        OSError: `AIRTABLE_API_KEY` environment is not set.
    """

    def __init__(self, url: str) -> None:
        airtable_api_key = os.getenv("AIRTABLE_API_KEY")
        if airtable_api_key is None:
            raise OSError("AIRTABLE_API_KEY environment is not set.")

        self.url = url
        self.headers = {
            "Authorization": f"Bearer {airtable_api_key}",
            "Content-Type": "application/json",
        }

    def add_records(self, records: List[Dict[str, Any]]) -> bool:
        """
        Add records to AirTable table.

        Args:
            records (List[Dict[str, Any]]):
                List of records in AirTable format.

        Returns:
            bool:
                Whether upload was a success.
        """
        try:
            response = requests.post(
                self.url, headers=self.headers, data=json.dumps({"records": records})
            )
        except Exception:
            return False

        return True if response.ok else False

    def batch_add_records(self, records: List[Dict[str, Any]]) -> bool:
        """
        Allow batching of record addition due to 10-element limit, then push.

        Args:
            records (List[Dict[str, Any]]):
                List of records in AirTable format.

        Returns:
            bool:
                Whether upload was a success.
        """
        BATCH_SIZE = 10
        for idx in range(0, len(records), BATCH_SIZE):
            batch = records[idx : idx + BATCH_SIZE]
            success = self.add_records(batch)
            if not success:
                return success
        return True

add_records(self, records)

Add records to AirTable table.

Parameters:

Name	Type	Description	Default
records	List[Dict[str, Any]]	List of records in AirTable format.	required

Returns:

Type	Description
bool	Whether upload was a success.

Source code in speechline/utils/airtable.py

def add_records(self, records: List[Dict[str, Any]]) -> bool:
    """
    Add records to AirTable table.

    Args:
        records (List[Dict[str, Any]]):
            List of records in AirTable format.

    Returns:
        bool:
            Whether upload was a success.
    """
    try:
        response = requests.post(
            self.url, headers=self.headers, data=json.dumps({"records": records})
        )
    except Exception:
        return False

    return True if response.ok else False

batch_add_records(self, records)

Allow batching of record addition due to 10-element limit, then push.

Parameters:

Name	Type	Description	Default
records	List[Dict[str, Any]]	List of records in AirTable format.	required

Returns:

Type	Description
bool	Whether upload was a success.

Source code in speechline/utils/airtable.py

def batch_add_records(self, records: List[Dict[str, Any]]) -> bool:
    """
    Allow batching of record addition due to 10-element limit, then push.

    Args:
        records (List[Dict[str, Any]]):
            List of records in AirTable format.

    Returns:
        bool:
            Whether upload was a success.
    """
    BATCH_SIZE = 10
    for idx in range(0, len(records), BATCH_SIZE):
        batch = records[idx : idx + BATCH_SIZE]
        success = self.add_records(batch)
        if not success:
            return success
    return True