codekingpro/portable-devtools

1from typing import Any, Iterator2 3from langchain_core.document_loaders import BaseLoader4from langchain_core.documents import Document5 6 7class AirtableLoader(BaseLoader):8    """Load the `Airtable` tables."""9 10    def __init__(11        self, api_token: str, table_id: str, base_id: str, **kwargs: Any12    ) -> None:13        """Initialize with API token and the IDs for table and base.14 15        Args:16            api_token: Airtable API token.17            table_id: Airtable table ID.18            base_id:19            kwargs: Additional parameters to pass to Table.all(). Refer to the20                pyairtable documentation for available options:21                https://pyairtable.readthedocs.io/en/latest/api.html#pyairtable.Table.all22        """  # noqa: E50123        self.api_token = api_token24        self.table_id = table_id25        self.base_id = base_id26        self.kwargs = kwargs27 28    def lazy_load(self) -> Iterator[Document]:29        """Lazy load Documents from table."""30 31        from pyairtable import Table32 33        table = Table(self.api_token, self.base_id, self.table_id)34        records = table.all(**self.kwargs)35        for record in records:36            metadata = {37                "source": self.base_id + "_" + self.table_id,38                "base_id": self.base_id,39                "table_id": self.table_id,40            }41            if "view" in self.kwargs:42                metadata["view"] = self.kwargs["view"]43            # Need to convert record from dict to str44            yield Document(page_content=str(record), metadata=metadata)45