Add personal access tokens support

.github/workflows/test.yml

@@ -0,0 +1,42 @@
+---
+name: Test
+
+on:
+  push:
+    branches: [main, master]
+  pull_request:
+    branches: [main, master]
+
+jobs:
+  test:
+    # if ITV fork
+    if: github.repository_owner == 'ITV'
+    runs-on: dev-target-airtable
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Set up Python 3.10
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+          cache: 'pip'
+
+      - name: Install Poetry
+        run: pip install poetry
+
+      - name: Cache Poetry dependencies
+        uses: actions/cache@v3
+        with:
+          path: |
+            ~/.cache/pypoetry
+            .venv
+          key: ${{ runner.os }}-poetry-${{ hashFiles('**/poetry.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-poetry-
+
+      - name: Install dependencies
+        run: poetry install
+
+      - name: Run tests
+        run: poetry run pytest

.gitignore

@@ -1,5 +1,6 @@
 # Secrets and internal config files
 .secrets/*
+.envrc
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

pyproject.toml

@@ -6,7 +6,7 @@ authors = ["hotglue"]
 license = "Apache 2.0"
 
 [tool.poetry.dependencies]
-python = "<3.11,>=3.7"
+python = ">=3.7,<3.11"
 requests = "^2.25.1"
 singer-sdk = "^0.3.2"
 airtable-python = "^0.1.1"

target_airtable/sinks.py

@@ -19,12 +19,19 @@ class AirtableSink(BatchSink):
     max_size = 10
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.client = Client(
-            self.config["client_id"],
-            self.config["client_secret"],
-            self.config["redirect_uri"],
-            uuid.uuid4().__str__().replace("-", "")*2
-        )
+        # Default to OAuth for backward compatibility
+        self.auth_method = self.config.get("authorization_method", "oauth")
+
+        if self.auth_method == "oauth":
+            self.client = Client(
+                self.config["client_id"],
+                self.config["client_secret"],
+                self.config.get("redirect_uri", ""),
+                uuid.uuid4().__str__().replace("-", "")*2
+            )
+        else:
+            # For personal access token, we don't need the OAuth client
+            self.client = None
 
     def _chunk(self, lst, n):
         """Yield successive n-sized chunks from lst."""
@@ -57,6 +64,9 @@ def gen_new_token(self, code):
 
     def _refresh_token(self):
         """Refresh OAuth token."""
+        if self.auth_method != "oauth":
+            return
+
         self.client.set_token({
             "access_token": self.config["access_token"],
             "refresh_token": self.config["refresh_token"],
@@ -75,7 +85,10 @@ def validate_response(self, response: requests.Response) -> None:
                 raise FatalAPIError(f"Airtable API Error: {response.text}")
 
         if response.status_code == 401:
-            self._refresh_token()
+            if self.auth_method == "oauth":
+                self._refresh_token()
+            else:
+                raise FatalAPIError(f"Authentication failed. Please check your personal access token.")
 
         if response.status_code == 429:
             raise RetriableAPIError(f"Too Many Requests for path: {response.request.url}")
@@ -102,7 +115,12 @@ def validate_response(self, response: requests.Response) -> None:
 
     @backoff.on_exception(backoff.expo, (requests.exceptions.RequestException, RetriableAPIError), max_tries=5, base=5, jitter=None)
     def _request(self, method, url, params=None, headers={}, data={}, *args, **kwargs):
-        new_headers = {'Authorization': 'Bearer {}'.format(self.config['access_token'])}
+        if self.auth_method == "oauth":
+            token = self.config['access_token']
+        else:
+            token = self.config['personal_access_token']
+
+        new_headers = {'Authorization': 'Bearer {}'.format(token)}
         headers.update(new_headers)
         response = requests.request(method, url, params=params, headers=headers, data=data, *args, **kwargs)
         return self.validate_response(response)

target_airtable/target.py

@@ -1,30 +1,89 @@
 """Airtable target class."""
 
-from pathlib import Path
-from typing import List
-
 from singer_sdk.target_base import Target
-from singer_sdk.sinks import Sink
 from singer_sdk import typing as th
+from singer_sdk.exceptions import ConfigValidationError
+
+from target_airtable.sinks import AirtableSink
 
-from target_airtable.sinks import (
-    AirtableSink,
-)
+# Authorization method constants
+AUTH_OAUTH = "oauth"
+AUTH_PERSONAL_ACCESS_TOKEN = "personal_access_token"
+VALID_AUTH_METHODS = [AUTH_OAUTH, AUTH_PERSONAL_ACCESS_TOKEN]
+
+# Required fields for each auth method
+OAUTH_FIELDS = ["access_token", "refresh_token", "client_id", "client_secret"]
 
 
 class TargetAirtable(Target):
     """Sample target for Airtable."""
 
     name = "target-airtable"
     config_jsonschema = th.PropertiesList(
-        th.Property("access_token", th.StringType, required=True),
-        th.Property("refresh_token", th.StringType, required=True),
-        th.Property("client_id", th.StringType, required=True),
-        th.Property("client_secret", th.StringType, required=True),
+        th.Property(
+            "authorization_method",
+            th.StringType,
+            required=False,
+            default="oauth",
+            description="Authorization method to use: 'oauth' or 'personal_access_token' (defaults to 'oauth' for backward compatibility)"
+        ),
+        th.Property("access_token", th.StringType, required=False),
+        th.Property("refresh_token", th.StringType, required=False),
+        th.Property("client_id", th.StringType, required=False),
+        th.Property("client_secret", th.StringType, required=False),
+        th.Property("personal_access_token", th.StringType, required=False),
         th.Property("base_id", th.StringType, required=True),
         th.Property("records_url", th.StringType)
     ).to_dict()
     default_sink_class = AirtableSink
 
+    def _validate_config(self, *, raise_errors: bool = True) -> tuple:
+        """Validate configuration with custom auth method checks."""
+        super()._validate_config(raise_errors=False)
+
+        errors = []
+        warnings = []
+
+        # Default to OAuth for backward compatibility
+        auth_method = self.config.get("authorization_method", AUTH_OAUTH)
+
+        # Validate auth_method value
+        if auth_method not in VALID_AUTH_METHODS:
+            errors.append(
+                f"authorization_method must be '{AUTH_OAUTH}' or '{AUTH_PERSONAL_ACCESS_TOKEN}', "
+                f"got '{auth_method}'"
+            )
+
+        if auth_method == AUTH_OAUTH:
+            self._validate_oauth_config(errors)
+        elif auth_method == AUTH_PERSONAL_ACCESS_TOKEN:
+            self._validate_pat_config(errors)
+
+        if raise_errors and errors:
+            raise ConfigValidationError(f"Config validation failed: {'; '.join(errors)}")
+
+        return errors, warnings
+
+    def _validate_oauth_config(self, errors: list) -> None:
+        """Validate OAuth-specific configuration."""
+        missing_fields = [field for field in OAUTH_FIELDS if not self.config.get(field)]
+        if missing_fields:
+            errors.append(f"OAuth authorization requires: {', '.join(missing_fields)}")
+
+        if self.config.get("personal_access_token"):
+            errors.append("personal_access_token should not be set when using OAuth authorization")
+
+    def _validate_pat_config(self, errors: list) -> None:
+        """Validate Personal Access Token configuration."""
+        if not self.config.get("personal_access_token"):
+            errors.append("personal_access_token is required when using personal_access_token authorization")
+
+        set_oauth_fields = [field for field in OAUTH_FIELDS if self.config.get(field)]
+        if set_oauth_fields:
+            errors.append(
+                f"OAuth fields ({', '.join(set_oauth_fields)}) should not be set "
+                f"when using personal_access_token authorization"
+            )
+
 if __name__ == '__main__':
     TargetAirtable.cli()