feat : Airtable connector Support

docs/connectors.md

@@ -187,3 +187,32 @@ yahoo_connector = YahooFinanceConnector("MSFT")
 df = SmartDataframe(yahoo_connector)
 df.chat("What is the closing price for yesterday?")
 ```
+
+## Airtable Connector
+
+The Airtable connector allows you to connect to Airtable Projects Tables, by simply passing the `base_id` , `api_key` and `table_name` of the table you want to analyze.
+
+To use the Airtable connector, you only need to import it into your Python code and pass it to a `SmartDataframe` or `SmartDatalake` object:
+
+```python
+from pandasai.connectors import AirtableConnector
+from pandasai import SmartDataframe
+
+
+airtable_connectors = AirtableConnector(
+    config={
+        "api_key": "AIRTABLE_API_TOKEN",
+        "table":"AIRTABLE_TABLE_NAME",
+        "base_id":"AIRTABLE_BASE_ID",
+        "where" : [
+            # this is optional and filters the data to
+            # reduce the size of the dataframe
+            ["Status" ,"=","In progress"]
+        ]
+    }
+)
+
+df = SmartDataframe(airtable_connectors)
+
+df.chat("How many rows are there in data ?")
+```

examples/from_airtable.py

@@ -0,0 +1,23 @@
+from pandasai.connectors import AirtableConnector
+from pandasai.llm import OpenAI
+from pandasai import SmartDataframe
+
+
+airtable_connectors = AirtableConnector(
+    config={
+        "api_key": "AIRTABLE_API_TOKEN",
+        "table": "AIRTABLE_TABLE_NAME",
+        "base_id": "AIRTABLE_BASE_ID",
+        "where": [
+            # this is optional and filters the data to
+            # reduce the size of the dataframe
+            ["Status", "=", "In progress"]
+        ],
+    }
+)
+
+llm = OpenAI("OPENAI_API_KEY")
+df = SmartDataframe(airtable_connectors, config={"llm": llm})
+
+response = df.chat("How many rows are there in data ?")
+print(response)

pandasai/connectors/__init__.py

@@ -9,6 +9,7 @@
 from .snowflake import SnowFlakeConnector
 from .databricks import DatabricksConnector
 from .yahoo_finance import YahooFinanceConnector
+from .airtable import AirtableConnector
 
 __all__ = [
     "BaseConnector",
@@ -18,4 +19,5 @@
     "YahooFinanceConnector",
     "SnowFlakeConnector",
     "DatabricksConnector",
+    "AirtableConnector",
 ]

pandasai/connectors/airtable.py

@@ -0,0 +1,292 @@
+"""
+Airtable connectors are used to connect airtable records.
+"""
+
+from .base import AirtableConnectorConfig, BaseConnector, BaseConnectorConfig
+from typing import Union, Optional
+import requests
+import pandas as pd
+import os
+from ..helpers.path import find_project_root
+import time
+import hashlib
+from ..exceptions import InvalidRequestError
+from functools import cache, cached_property
+
+
+class AirtableConnector(BaseConnector):
+    """
+    Airtable connector to retrieving record data.
+    """
+
+    _rows_count: int = None
+    _columns_count: int = None
+    _instance = None
+
+    def __init__(
+        self,
+        config: Optional[Union[AirtableConnectorConfig, dict]] = None,
+        cache_interval: int = 600,
+    ):
+        if isinstance(config, dict):
+            if "api_key" in config and "base_id" in config and "table" in config:
+                config = AirtableConnectorConfig(**config)
+            else:
+                raise KeyError(
+                    "Please specify all api_key,table,base_id properly in config ."
+                )
+
+        elif not config:
+            airtable_env_vars = {
+                "api_key": "AIRTABLE_API_TOKEN",
+                "base_id": "AIRTABLE_BASE_ID",
+                "table": "AIRTABLE_TABLE_NAME",
+            }
+            config = AirtableConnectorConfig(
+                **self._populate_config_from_env(config, airtable_env_vars)
+            )
+
+        self._root_url: str = "https://api.airtable.com/v0/"
+        self._cache_interval = cache_interval
+
+        super().__init__(config)
+
+    def _init_connection(self, config: BaseConnectorConfig):
+        """
+        make connection to database
+        """
+        config = config.dict()
+        url = f"{self._root_url}{config['base_id']}/{config['table']}"
+        response = requests.head(
+            url=url, headers={"Authorization": f"Bearer {config['api_key']}"}
+        )
+        if response.status_code == 200:
+            self.logger.log(
+                """
+                Connected to Airtable.
+            """
+            )
+        else:
+            raise InvalidRequestError(
+                f"""Failed to connect to Airtable. 
+                    Status code: {response.status_code}, 
+                    message: {response.text}"""
+            )
+
+    def _get_cache_path(self, include_additional_filters: bool = False):
+        """
+        Return the path of the cache file.
+
+        Returns :
+            str : The path of the cache file.
+        """
+        cache_dir = os.path.join(os.getcwd(), "")
+        try:
+            cache_dir = os.path.join((find_project_root()), "cache")
+        except ValueError:
+            cache_dir = os.path.join(os.getcwd(), "cache")
+        return os.path.join(cache_dir, f"{self._config.table}_data.parquet")
+
+    def _cached(self, include_additional_filters: bool = False):
+        """
+        Returns the cached Airtable data if it exists and
+        is not older than the cache interval.
+
+        Returns :
+            DataFrame | None : The cached data if
+                it exists and is not older than the cache
+                interval, None otherwise.
+        """
+        cache_path = self._get_cache_path(include_additional_filters)
+        if not os.path.exists(cache_path):
+            return None
+
+        # If the file is older than 1 day , delete it.
+        if os.path.getmtime(cache_path) < time.time() - self._cache_interval:
+            if self.logger:
+                self.logger.log(f"Deleting expired cached data from {cache_path}")
+            os.remove(cache_path)
+            return None
+
+        if self.logger:
+            self.logger.log(f"Loading cached data from {cache_path}")
+
+        return cache_path
+
+    def _save_cache(self, df):
+        """
+        Save the given DataFrame to the cache.
+
+        Args:
+            df (DataFrame): The DataFrame to save to the cache.
+        """
+        filename = self._get_cache_path(
+            include_additional_filters=self._additional_filters is not None
+            and len(self._additional_filters) > 0
+        )
+        df.to_parquet(filename)
+
+    @property
+    def fallback_name(self):
+        """
+        Returns the fallback table name of the connector.
+
+        Returns :
+            str : The fallback table name of the connector.
+        """
+        return self._config.table
+
+    def execute(self):
+        """
+        Execute the connector and return the result.
+
+        Returns:
+            DataFrameType: The result of the connector.
+        """
+        cached = self._cached() or self._cached(include_additional_filters=True)
+        if cached:
+            return pd.read_parquet(cached)
+
+        if isinstance(self._instance, pd.DataFrame):
+            return self._instance
+        else:
+            self._instance = self._fetch_data()
+
+        return self._instance
+
+    def _build_formula(self):
+        """
+        Build Airtable query formula for filtering.
+        """
+
+        condition_strings = []
+        if self._config.where is not None:
+            for i in self._config.where:
+                filter_query = f"{i[0]}{i[1]}'{i[2]}'"
+                condition_strings.append(filter_query)
+        filter_formula = f'AND({",".join(condition_strings)})'
+        return filter_formula
+
+    def _request_api(self, params):
+        url = f"{self._root_url}{self._config.base_id}/{self._config.table}"
+        response = requests.get(
+            url=url,
+            headers={"Authorization": f"Bearer {self._config.api_key}"},
+            params=params,
+        )
+        return response
+
+    def _fetch_data(self):
+        """
+        Feteches data from airtable server through
+            API and converts it to DataFrame.
+        """
+
+        params = {}
+        if self._config.where is not None:
+            params["filterByFormula"] = self._build_formula()
+
+        params["pageSize"] = 100
+        params["offset"] = "0"
+
+        data = []
+        while True:
+            response = self._request_api(params=params)
+
+            if response.status_code == 200:
+                res = response.json()
+                data.append(res)
+                if len(res["records"]) < 100:
+                    break
+            else:
+                raise InvalidRequestError(
+                    f"""Failed to connect to Airtable. 
+                        Status code: {response.status_code}, 
+                        message: {response.text}"""
+                )
+
+            if "offset" in res:
+                params["offset"] = res["offset"]
+
+        data = self._preprocess(data=data)
+        return data
+
+    def _preprocess(self, data):
+        """
+        Preprocesses Json response data
+        To prepare dataframe correctly.
+        """
+        columns = set()
+        data_dict_list = []
+        for item in data:
+            for entry in item["records"]:
+                data_dict = {"id": entry["id"], "createdTime": entry["createdTime"]}
+                for field_name, field_value in entry["fields"].items():
+                    data_dict[field_name] = field_value
+                    columns.add(field_name)
+                data_dict_list.append(data_dict)
+
+        df = pd.DataFrame(data_dict_list)
+        return df
+
+    @cache
+    def head(self):
+        """
+        Return the head of the table that
+          the connector is connected to.
+
+        Returns :
+            DatFrameType: The head of the data source
+                 that the conector is connected to .
+        """
+        data = self._request_api(params={"maxRecords": 5})
+        data = self._preprocess([data.json()])
+        return data
+
+    @cached_property
+    def rows_count(self):
+        """
+        Return the number of rows in the data source that the connector is
+        connected to.
+
+        Returns:
+            int: The number of rows in the data source that the connector is
+            connected to.
+        """
+        if self._rows_count is not None:
+            return self._rows_count
+        data = self.execute()
+        self._rows_count = len(data)
+        return self._rows_count
+
+    @cached_property
+    def columns_count(self):
+        """
+        Return the number of columns in the data source that the connector is
+        connected to.
+
+        Returns:
+            int: The number of columns in the data source that the connector is
+            connected to.
+        """
+        if self._columns_count is not None:
+            return self._columns_count
+        data = self.head()
+        self._columns_count = len(data.columns)
+        return self._columns_count
+
+    @property
+    def column_hash(self):
+        """
+        Return the hash code that is unique to the columns of the data source
+        that the connector is connected to.
+
+        Returns:
+            int: The hash code that is unique to the columns of the data source
+            that the connector is connected to.
+        """
+        if not isinstance(self._instance, pd.DataFrame):
+            self._instance = self.execute()
+        columns_str = "|".join(self._instance.columns)
+        columns_str += "WHERE" + self._build_formula()
+        return hashlib.sha256(columns_str.encode("utf-8")).hexdigest()

pandasai/connectors/base.py

@@ -20,6 +20,16 @@ class BaseConnectorConfig(BaseModel):
     where: list[list[str]] = None
 
 
+class AirtableConnectorConfig(BaseConnectorConfig):
+    """
+    Connecter configuration for Airtable data.
+    """
+
+    api_key: str
+    base_id: str
+    database: str = "airtable_data"
+
+
 class SQLBaseConnectorConfig(BaseConnectorConfig):
     """
     Base Connector configuration.

pandasai/exceptions.py

@@ -5,6 +5,16 @@
 """
 
 
+class InvalidRequestError(Exception):
+
+    """
+    Raised when the request is not succesfull.
+
+    Args :
+        Exception (Exception): InvalidRequestError
+    """
+
+
 class APIKeyNotFoundError(Exception):
 
     """