feat: OAuth flow

tap_linkedin_ads/__main__.py

@@ -0,0 +1,7 @@
+"""LinkedInAds entry point."""
+
+from __future__ import annotations
+
+from tap_linkedin_ads.tap import TapLinkedInAds
+
+TapLinkedInAds.cli()

tap_linkedin_ads/client.py

@@ -7,15 +7,32 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
-from singer_sdk.authenticators import BearerTokenAuthenticator
+import requests
+from singer_sdk.authenticators import (
+    BearerTokenAuthenticator,
+    OAuthAuthenticator,
+    SingletonMeta,
+)
 from singer_sdk.streams import RESTStream
 
-if t.TYPE_CHECKING:
-    import requests
-
 SCHEMAS_DIR = Path(__file__).parent / Path("./schemas")
 UTC = timezone.utc
 
+_Auth = t.Callable[[requests.PreparedRequest], requests.PreparedRequest]
+
+
+class LinkedInAdsOAuthAuthenticator(OAuthAuthenticator, metaclass=SingletonMeta):
+    """Authenticator class for LinkedInAds."""
+
+    @property
+    def oauth_request_body(self):
+        return {
+            "grant_type": "refresh_token",
+            "client_id": self.config["oauth_credentials"]["client_id"],
+            "client_secret": self.config["oauth_credentials"]["client_secret"],
+            "refresh_token": self.config["oauth_credentials"]["refresh_token"],
+        }
+
 
 class LinkedInAdsStream(RESTStream):
     """LinkedInAds stream class."""
@@ -26,12 +43,17 @@ class LinkedInAdsStream(RESTStream):
     )
 
     @property
-    def authenticator(self) -> BearerTokenAuthenticator:
+    def authenticator(self) -> _Auth:
         """Return a new authenticator object.
 
         Returns:
             An authenticator instance.
         """
+        if "oauth_credentials" in self.config:
+            return LinkedInAdsOAuthAuthenticator(
+                self,
+                auth_endpoint="https://www.linkedin.com/oauth/v2/accessToken",
+            )
         return BearerTokenAuthenticator.create_for_stream(
             self,
             token=self.config["access_token"],

tap_linkedin_ads/streams.py

@@ -1434,8 +1434,7 @@ def post_process(self, row: dict, context: dict | None = None) -> dict | None:
                 "%Y-%m-%d",
             ).astimezone(UTC)
 
-        with contextlib.suppress(IndexError):
-            row["creative_id"] = self.config["creative"]
+        row["creative_id"] = self.config["creative"]
 
         viral_registrations = row.pop("viralRegistrations", None)
         if viral_registrations:

tap_linkedin_ads/tap.py

@@ -25,9 +25,30 @@ class TapLinkedInAds(Tap):
         th.Property(
             "access_token",
             th.StringType,
-            required=True,
             description="The token to authenticate against the API service",
         ),
+        # OAuth
+        th.Property(
+            "oauth_credentials",
+            th.ObjectType(
+                th.Property(
+                    "refresh_token",
+                    th.StringType,
+                    description="LinkedIn Ads Refresh Token",
+                ),
+                th.Property(
+                    "client_id",
+                    th.StringType,
+                    description="LinkedIn Ads Client ID",
+                ),
+                th.Property(
+                    "client_secret",
+                    th.StringType,
+                    description="LinkedIn Ads Client Secret",
+                ),
+            ),
+            description="LinkedIn Ads OAuth Credentials",
+        ),
         th.Property(
             "start_date",
             th.DateTimeType,