AirTable Interface

audio_recording_logger.airtable

AirTable

Source code in src/audio_recording_logger/airtable.py

class AirTable:
    def __init__(self, url: str) -> None:
        """Constructor for AirTable table.

        Args:
            url (str): URL of AirTable table.
        """
        self.url = url
        self.headers = {
            "Authorization": f"Bearer {os.environ['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 as exc:
            print(exc)
            return False
        else:
            if response.ok:
                return True
            else:
                print(f"Failed to patch {records}")
                return 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

__init__(url)

Constructor for AirTable table.

Parameters:

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

Source code in src/audio_recording_logger/airtable.py

def __init__(self, url: str) -> None:
    """Constructor for AirTable table.

    Args:
        url (str): URL of AirTable table.
    """
    self.url = url
    self.headers = {
        "Authorization": f"Bearer {os.environ['AIRTABLE_API_KEY']}",
        "Content-Type": "application/json",
    }

add_records(records)

Add records to AirTable table.

Parameters:

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

Returns:

Name	Type	Description
bool	bool	Whether upload was a success.

Source code in src/audio_recording_logger/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 as exc:
        print(exc)
        return False
    else:
        if response.ok:
            return True
        else:
            print(f"Failed to patch {records}")
            return False

batch_add_records(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:

Name	Type	Description
bool	bool	Whether upload was a success.

Source code in src/audio_recording_logger/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