Low-Code: added SessionTokenAuthenticator (#19716)

* added SessionTokenAuthenticator for low-code

* added doc comments and formatted files

* fixed doc strings

* added login_url and validate_session_url to config

* removed unused f-string, formatted

* Update log

Co-authored-by: Serhii Lazebnyi <[email protected]>

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.13.1
+Low-code: Add `SessionTokenAuthenticator`
+
 ## 0.13.0
 Add `Stream.check_availability` and `Stream.AvailabilityStrategy`. Make `HttpAvailabilityStrategy` the default `HttpStream.AvailabilityStrategy`.
 

airbyte-cdk/python/airbyte_cdk/sources/declarative/auth/token.py

@@ -3,13 +3,16 @@
 #
 
 import base64
+import logging
 from dataclasses import InitVar, dataclass
 from typing import Any, Mapping, Union
 
+import requests
 from airbyte_cdk.sources.declarative.auth.declarative_authenticator import DeclarativeAuthenticator
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.types import Config
 from airbyte_cdk.sources.streams.http.requests_native_auth.abstract_token import AbstractHeaderAuthenticator
+from cachetools import TTLCache, cached
 from dataclasses_jsonschema import JsonSchemaMixin
 
 
@@ -115,3 +118,125 @@ def token(self) -> str:
         auth_string = f"{self._username.eval(self.config)}:{self._password.eval(self.config)}".encode("utf8")
         b64_encoded = base64.b64encode(auth_string).decode("utf8")
         return f"Basic {b64_encoded}"
+
+
+"""
+    maxsize - The maximum size of the cache
+    ttl - time-to-live value in seconds
+    docs https://cachetools.readthedocs.io/en/latest/
+    maxsize=1000 - when the cache is full, in this case more than 1000,
+    i.e. by adding another item the cache would exceed its maximum size, the cache must choose which item(s) to discard
+    ttl=86400 means that cached token will live for 86400 seconds (one day)
+"""
+cacheSessionTokenAuthenticator = TTLCache(maxsize=1000, ttl=86400)
+
+
+@cached(cacheSessionTokenAuthenticator)
+def get_new_session_token(api_url: str, username: str, password: str, response_key: str) -> str:
+    """
+    This method retrieves session token from api by username and password for SessionTokenAuthenticator.
+    It's cashed to avoid a multiple calling by sync and updating session token every stream sync.
+    Args:
+        api_url: api url for getting new session token
+        username: username for auth
+        password: password for auth
+        response_key: field name in response to retrieve a session token
+
+    Returns:
+        session token
+    """
+    response = requests.post(
+        f"{api_url}",
+        headers={"Content-Type": "application/json"},
+        json={"username": username, "password": password},
+    )
+    response.raise_for_status()
+    if not response.ok:
+        raise ConnectionError(f"Failed to retrieve new session token, response code {response.status_code} because {response.reason}")
+    return response.json()[response_key]
+
+
+@dataclass
+class SessionTokenAuthenticator(AbstractHeaderAuthenticator, DeclarativeAuthenticator, JsonSchemaMixin):
+    """
+    Builds auth based on session tokens.
+    A session token is a random value generated by a server to identify
+    a specific user for the duration of one interaction session.
+
+    The header is of the form
+    `"Specific Header": "Session Token Value"`
+
+    Attributes:
+        api_url (Union[InterpolatedString, str]): Base api url of source
+        username (Union[InterpolatedString, str]): The username
+        config (Config): The user-provided configuration as specified by the source's spec
+        password (Union[InterpolatedString, str]): The password
+        header (Union[InterpolatedString, str]): Specific header of source for providing session token
+        options (Mapping[str, Any]): Additional runtime parameters to be used for string interpolation
+        session_token (Union[InterpolatedString, str]): Session token generated by user
+        session_token_response_key (Union[InterpolatedString, str]): Key for retrieving session token from api response
+        login_url (Union[InterpolatedString, str]): Url fot getting a specific session token
+        validate_session_url (Union[InterpolatedString, str]): Url to validate passed session token
+    """
+
+    api_url: Union[InterpolatedString, str]
+    header: Union[InterpolatedString, str]
+    session_token: Union[InterpolatedString, str]
+    session_token_response_key: Union[InterpolatedString, str]
+    username: Union[InterpolatedString, str]
+    config: Config
+    options: InitVar[Mapping[str, Any]]
+    login_url: Union[InterpolatedString, str]
+    validate_session_url: Union[InterpolatedString, str]
+    password: Union[InterpolatedString, str] = ""
+
+    def __post_init__(self, options):
+        self._username = InterpolatedString.create(self.username, options=options)
+        self._password = InterpolatedString.create(self.password, options=options)
+        self._api_url = InterpolatedString.create(self.api_url, options=options)
+        self._header = InterpolatedString.create(self.header, options=options)
+        self._session_token = InterpolatedString.create(self.session_token, options=options)
+        self._session_token_response_key = InterpolatedString.create(self.session_token_response_key, options=options)
+        self._login_url = InterpolatedString.create(self.login_url, options=options)
+        self._validate_session_url = InterpolatedString.create(self.validate_session_url, options=options)
+
+        self.logger = logging.getLogger("airbyte")
+
+    @property
+    def auth_header(self) -> str:
+        return self._header.eval(self.config)
+
+    @property
+    def token(self) -> str:
+        if self._session_token.eval(self.config):
+            if self.is_valid_session_token():
+                return self._session_token.eval(self.config)
+        if self._password.eval(self.config) and self._username.eval(self.config):
+            username = self._username.eval(self.config)
+            password = self._password.eval(self.config)
+            session_token_response_key = self._session_token_response_key.eval(self.config)
+            api_url = f"{self._api_url.eval(self.config)}{self._login_url.eval(self.config)}"
+
+            self.logger.info("Using generated session token by username and password")
+            return get_new_session_token(api_url, username, password, session_token_response_key)
+
+        raise ConnectionError("Invalid credentials: session token is not valid or provide username and password")
+
+    def is_valid_session_token(self) -> bool:
+        try:
+            response = requests.get(
+                f"{self._api_url.eval(self.config)}{self._validate_session_url.eval(self.config)}",
+                headers={self.auth_header: self._session_token.eval(self.config)},
+            )
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            if e.response.status_code == requests.codes["unauthorized"]:
+                self.logger.info(f"Unable to connect by session token from config due to {str(e)}")
+                return False
+            else:
+                raise ConnectionError(f"Error while validating session token: {e}")
+        if response.ok:
+            self.logger.info("Connection check for source is successful.")
+            return True
+        else:
+            raise ConnectionError(f"Failed to retrieve new session token, response code {response.status_code} because {response.reason}")

airbyte-cdk/python/airbyte_cdk/sources/declarative/config_component_schema.json

@@ -289,6 +289,9 @@
                 },
                 {
                   "$ref": "#/definitions/BasicHttpAuthenticator"
+                },
+                {
+                  "$ref": "#/definitions/SessionTokenAuthenticator"
                 }
               ]
             },
@@ -629,6 +632,75 @@
       ],
       "description": "\n    Builds auth based off the basic authentication scheme as defined by RFC 7617, which transmits credentials as USER ID/password pairs, encoded using base64\n    https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication#basic_authentication_scheme\n\n    The header is of the form\n    `\"Authorization\": \"Basic <encoded_credentials>\"`\n\n    Attributes:\n        username (Union[InterpolatedString, str]): The username\n        config (Config): The user-provided configuration as specified by the source's spec\n        password (Union[InterpolatedString, str]): The password\n        options (Mapping[str, Any]): Additional runtime parameters to be used for string interpolation\n    "
     },
+    "SessionTokenAuthenticator": {
+      "allOf": [
+        {
+          "$ref": "#/definitions/DeclarativeAuthenticator"
+        },
+        {
+          "type": "object",
+          "required": ["api_url", "header"],
+          "properties": {
+            "api_url": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/InterpolatedString"
+                },
+                {
+                  "type": "string"
+                }
+              ]
+            },
+            "session_token": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/InterpolatedString"
+                },
+                {
+                  "type": "string"
+                }
+              ],
+              "default": ""
+            },
+            "username": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/InterpolatedString"
+                },
+                {
+                  "type": "string"
+                }
+              ],
+              "default": ""
+            },
+            "password": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/InterpolatedString"
+                },
+                {
+                  "type": "string"
+                }
+              ],
+              "default": ""
+            },
+            "header": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/InterpolatedString"
+                },
+                {
+                  "type": "string"
+                }
+              ]
+            },
+            "config": {
+              "type": "object"
+            }
+          }
+        }
+      ]
+    },
     "CompositeErrorHandler": {
       "allOf": [
         {

airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/class_types_registry.py

@@ -6,7 +6,12 @@
 
 from airbyte_cdk.sources.declarative.auth.declarative_authenticator import NoAuth
 from airbyte_cdk.sources.declarative.auth.oauth import DeclarativeOauth2Authenticator
-from airbyte_cdk.sources.declarative.auth.token import ApiKeyAuthenticator, BasicHttpAuthenticator, BearerAuthenticator
+from airbyte_cdk.sources.declarative.auth.token import (
+    ApiKeyAuthenticator,
+    BasicHttpAuthenticator,
+    BearerAuthenticator,
+    SessionTokenAuthenticator,
+)
 from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.declarative_stream import DeclarativeStream
 from airbyte_cdk.sources.declarative.extractors.dpath_extractor import DpathExtractor
@@ -78,6 +83,7 @@
     "SingleSlice": SingleSlice,
     "Spec": Spec,
     "SubstreamSlicer": SubstreamSlicer,
+    "SessionTokenAuthenticator": SessionTokenAuthenticator,
     "WaitUntilTimeFromHeader": WaitUntilTimeFromHeaderBackoffStrategy,
     "WaitTimeFromHeader": WaitTimeFromHeaderBackoffStrategy,
 }

airbyte-cdk/python/setup.py

@@ -15,7 +15,7 @@
 
 setup(
     name="airbyte-cdk",
-    version="0.13.0",
+    version="0.13.1",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",
@@ -58,6 +58,7 @@
         "requests_cache",
         "Deprecated~=1.2",
         "Jinja2~=3.1.2",
+        "cachetools",
     ],
     python_requires=">=3.9",
     extras_require={