refactor(cdk): replace token_prefix with interpolation approach for SessionTokenAuthenticator

This replaces the token_prefix approach with a more flexible interpolation approach using api_token template. Users can now use Jinja templates like 'Token {{ session_token }}' for Django REST Framework APIs.

Changes:
- Replace PrefixedTokenProvider with InterpolatedSessionTokenProvider
- Update schema to use api_token field instead of token_prefix
- Default api_token is '{{ session_token }}' for backward compatibility
- Update factory to always wrap with InterpolatedSessionTokenProvider
- Update tests to reflect new approach

Co-Authored-By: Ryan Waskewich <ryan.waskewich@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/auth/token_provider.py

@@ -85,15 +85,21 @@ def get_token(self) -> str:
 
 
 @dataclass
-class PrefixedTokenProvider(TokenProvider):
-    """Wraps a TokenProvider and prepends a prefix to the token value.
+class InterpolatedSessionTokenProvider(TokenProvider):
+    """Provides a token by interpolating a template with the session token.
 
-    This is useful for APIs that require a specific prefix before the token,
-    such as Django REST Framework APIs that expect "Token <value>" format.
+    This allows flexible token formatting, such as "Token {{ session_token }}"
+    for Django REST Framework APIs that expect "Authorization: Token <value>".
     """
 
-    token_provider: TokenProvider
-    prefix: str
+    config: Config
+    api_token: Union[InterpolatedString, str]
+    session_token_provider: TokenProvider
+    parameters: Mapping[str, Any]
+
+    def __post_init__(self) -> None:
+        self._token_template = InterpolatedString.create(self.api_token, parameters=self.parameters)
 
     def get_token(self) -> str:
-        return f"{self.prefix}{self.token_provider.get_token()}"
+        session_token = self.session_token_provider.get_token()
+        return str(self._token_template.eval(self.config, session_token=session_token))

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2067,14 +2067,18 @@ definitions:
             field_name: Authorization
           - inject_into: request_parameter
             field_name: authKey
-      token_prefix:
-        title: Token Prefix
-        description: 'A prefix to prepend to the session token when injecting it into requests. For example, use "Token " (with trailing space) for APIs that expect "Authorization: Token <token>".'
+      api_token:
+        title: API Token Template
+        description: 'A template for the token value to inject. Use {{ session_token }} to reference the session token. For example, use "Token {{ session_token }}" for APIs that expect "Authorization: Token <token>".'
         type: string
-        default: ""
+        default: "{{ session_token }}"
+        interpolation_context:
+          - config
+          - session_token
         examples:
-          - "Token "
-          - "Bearer "
+          - "{{ session_token }}"
+          - "Token {{ session_token }}"
+          - "Bearer {{ session_token }}"
   SessionTokenRequestBearerAuthenticator:
     title: Bearer Authenticator
     description: Authenticator for requests using the session token as a standard bearer token.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -2055,11 +2057,15 @@ class SessionTokenRequestApiKeyAuthenticator(BaseModel):
         ],
         title="Inject API Key Into Outgoing HTTP Request",
     )
-    token_prefix: Optional[str] = Field(
-        "",
-        description='A prefix to prepend to the session token when injecting it into requests. For example, use "Token " (with trailing space) for APIs that expect "Authorization: Token <token>".',
-        examples=["Token ", "Bearer "],
-        title="Token Prefix",
+    api_token: Optional[str] = Field(
+        "{{ session_token }}",
+        description='A template for the token value to inject. Use {{ session_token }} to reference the session token. For example, use "Token {{ session_token }}" for APIs that expect "Authorization: Token <token>".',
+        examples=[
+            "{{ session_token }}",
+            "Token {{ session_token }}",
+            "Bearer {{ session_token }}",
+        ],
+        title="API Token Template",
     )
 
 
@@ -2745,7 +2751,7 @@ class HttpRequester(BaseModelWithDeprecations):
     )
     use_cache: Optional[bool] = Field(
         False,
-        description="Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).",
+        description="Enables stream requests caching. This field is automatically set by the CDK.",
         title="Use Cache",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -67,8 +67,8 @@
     LegacySessionTokenAuthenticator,
 )
 from airbyte_cdk.sources.declarative.auth.token_provider import (
+    InterpolatedSessionTokenProvider,
     InterpolatedStringTokenProvider,
-    PrefixedTokenProvider,
     SessionTokenProvider,
     TokenProvider,
 )
@@ -1170,14 +1170,16 @@ def create_session_token_authenticator(
                 token_provider=token_provider,
             )
         else:
-            # Get the token_prefix if specified, wrap the token provider if needed
-            token_prefix = getattr(model.request_authentication, "token_prefix", None) or ""
-            final_token_provider: TokenProvider = token_provider
-            if token_prefix:
-                final_token_provider = PrefixedTokenProvider(
-                    token_provider=token_provider,
-                    prefix=token_prefix,
-                )
+            # Get the api_token template if specified, default to just the session token
+            api_token_template = (
+                getattr(model.request_authentication, "api_token", None) or "{{ session_token }}"
+            )
+            final_token_provider: TokenProvider = InterpolatedSessionTokenProvider(
+                config=config,
+                api_token=api_token_template,
+                session_token_provider=token_provider,
+                parameters=model.parameters or {},
+            )
             return self.create_api_key_authenticator(
                 ApiKeyAuthenticatorModel(
                     type="ApiKeyAuthenticator",

unit_tests/sources/declarative/auth/test_token_provider.py

@@ -9,8 +9,8 @@
 from isodate import parse_duration
 
 from airbyte_cdk.sources.declarative.auth.token_provider import (
+    InterpolatedSessionTokenProvider,
     InterpolatedStringTokenProvider,
-    PrefixedTokenProvider,
     SessionTokenProvider,
 )
 from airbyte_cdk.sources.declarative.exceptions import ReadException
@@ -83,19 +83,26 @@ def test_session_token_provider_ignored_response():
 
 
 @pytest.mark.parametrize(
-    "prefix,expected_token",
+    "api_token_template,expected_token",
     [
-        pytest.param("Token ", "Token my_token", id="token_prefix"),
-        pytest.param("Bearer ", "Bearer my_token", id="bearer_prefix"),
-        pytest.param("", "my_token", id="empty_prefix"),
-        pytest.param("Custom-", "Custom-my_token", id="custom_prefix"),
+        pytest.param("Token {{ session_token }}", "Token my_token", id="token_prefix"),
+        pytest.param("Bearer {{ session_token }}", "Bearer my_token", id="bearer_prefix"),
+        pytest.param("{{ session_token }}", "my_token", id="just_session_token"),
+        pytest.param("Custom-{{ session_token }}", "Custom-my_token", id="custom_prefix"),
+        pytest.param(
+            "realm=xyz, token={{ session_token }}",
+            "realm=xyz, token=my_token",
+            id="complex_format",
+        ),
     ],
 )
-def test_prefixed_token_provider(prefix, expected_token):
-    """Test that PrefixedTokenProvider correctly prepends prefix to token."""
+def test_interpolated_session_token_provider(api_token_template, expected_token):
+    """Test that InterpolatedSessionTokenProvider correctly interpolates session token."""
     underlying_provider = create_session_token_provider()
-    prefixed_provider = PrefixedTokenProvider(
-        token_provider=underlying_provider,
-        prefix=prefix,
+    interpolated_provider = InterpolatedSessionTokenProvider(
+        config={"some_config": "value"},
+        api_token=api_token_template,
+        session_token_provider=underlying_provider,
+        parameters={},
     )
-    assert prefixed_provider.get_token() == expected_token
+    assert interpolated_provider.get_token() == expected_token

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -43,7 +43,7 @@
     LegacySessionTokenAuthenticator,
 )
 from airbyte_cdk.sources.declarative.auth.token_provider import (
-    PrefixedTokenProvider,
+    InterpolatedSessionTokenProvider,
     SessionTokenProvider,
 )
 from airbyte_cdk.sources.declarative.checks import CheckStream
@@ -1844,22 +1844,25 @@ def test_create_request_with_session_authenticator():
     )
 
     assert isinstance(selector.authenticator, ApiKeyAuthenticator)
-    assert isinstance(selector.authenticator.token_provider, SessionTokenProvider)
-    assert selector.authenticator.token_provider.session_token_path == ["id"]
-    assert isinstance(selector.authenticator.token_provider.login_requester, HttpRequester)
-    assert selector.authenticator.token_provider.session_token_path == ["id"]
+    # Default behavior wraps with InterpolatedSessionTokenProvider using "{{ session_token }}"
+    assert isinstance(selector.authenticator.token_provider, InterpolatedSessionTokenProvider)
+    assert selector.authenticator.token_provider.api_token == "{{ session_token }}"
+    session_token_provider = selector.authenticator.token_provider.session_token_provider
+    assert isinstance(session_token_provider, SessionTokenProvider)
+    assert session_token_provider.session_token_path == ["id"]
+    assert isinstance(session_token_provider.login_requester, HttpRequester)
     assert (
-        selector.authenticator.token_provider.login_requester._url_base.eval(input_config)
+        session_token_provider.login_requester._url_base.eval(input_config)
         == "https://api.sendgrid.com"
     )
-    assert selector.authenticator.token_provider.login_requester.get_request_body_json() == {
+    assert session_token_provider.login_requester.get_request_body_json() == {
         "username": "lists",
         "password": "verysecrettoken",
     }
 
 
-def test_create_request_with_session_authenticator_with_token_prefix():
-    """Test that token_prefix wraps the token provider with PrefixedTokenProvider."""
+def test_create_request_with_session_authenticator_with_api_token_template():
+    """Test that api_token wraps the token provider with InterpolatedSessionTokenProvider."""
     content = """
 requester:
   type: HttpRequester
@@ -1888,7 +1891,7 @@ def test_create_request_with_session_authenticator_with_token_prefix():
         type: RequestOption
         field_name: Authorization
         inject_into: header
-      token_prefix: "Token "
+      api_token: "Token {{ session_token }}"
     """
     name = "name"
     parsed_manifest = YamlDeclarativeSource._parse(content)
@@ -1906,9 +1909,11 @@ def test_create_request_with_session_authenticator_with_token_prefix():
     )
 
     assert isinstance(selector.authenticator, ApiKeyAuthenticator)
-    assert isinstance(selector.authenticator.token_provider, PrefixedTokenProvider)
-    assert selector.authenticator.token_provider.prefix == "Token "
-    assert isinstance(selector.authenticator.token_provider.token_provider, SessionTokenProvider)
+    assert isinstance(selector.authenticator.token_provider, InterpolatedSessionTokenProvider)
+    assert selector.authenticator.token_provider.api_token == "Token {{ session_token }}"
+    assert isinstance(
+        selector.authenticator.token_provider.session_token_provider, SessionTokenProvider
+    )
 
 
 def test_given_composite_error_handler_does_not_match_response_then_fallback_on_default_error_handler(