An Airbyte source connector for extracting data from Webflow CMS collections
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Authentication classes for Webflow API integration, providing token-based authentication with proper API versioning headers required by Webflow's API.
Mixin class that adds Webflow-specific API version headers to authentication. Webflow requires an "accept-version" header for API compatibility.
class WebflowAuthMixin:
"""Mixin for adding Webflow API version headers to authentication."""
def __init__(self, *, accept_version_header: str = "accept-version", accept_version: str, **kwargs):
"""
Initialize Webflow authentication mixin.
Parameters:
- accept_version_header: Header name for API version (default: "accept-version")
- accept_version: API version string (e.g., "1.0.0")
- **kwargs: Additional arguments passed to parent class
"""
def get_auth_header(self) -> Mapping[str, Any]:
"""
Get authentication headers including API version.
Returns:
Dictionary with authentication headers plus accept-version header
"""Complete authenticator for Webflow API that combines token authentication with required API versioning headers.
class WebflowTokenAuthenticator(WebflowAuthMixin, TokenAuthenticator):
"""Token-based authenticator for Webflow API with version headers."""from source_webflow.auth import WebflowTokenAuthenticator
# Create authenticator with API token
auth = WebflowTokenAuthenticator(
token="your_webflow_api_token",
accept_version="1.0.0"
)
# The authenticator will automatically include both:
# - Authorization: Bearer your_webflow_api_token
# - accept-version: 1.0.0from source_webflow.auth import WebflowTokenAuthenticator
import requests
# Create authenticator
auth = WebflowTokenAuthenticator(
token="your_api_token",
accept_version="1.0.0"
)
# Get headers for manual requests
headers = auth.get_auth_header()
print(headers)
# Output: {
# 'Authorization': 'Bearer your_api_token',
# 'accept-version': '1.0.0'
# }
# Use with requests
response = requests.get(
"https://api.webflow.com/sites/your_site_id/collections",
headers=headers
)from source_webflow.auth import WebflowTokenAuthenticator
# Use custom version header name (uncommon)
auth = WebflowTokenAuthenticator(
token="your_api_token",
accept_version="1.0.0",
accept_version_header="api-version" # Custom header name
)Obtain a Webflow API token:
Webflow uses semantic versioning for their API:
Common authentication errors:
The authenticator itself doesn't handle these errors - they're returned by the API and should be handled at the stream or source level.
The WebflowTokenAuthenticator integrates seamlessly with Airbyte's CDK:
from airbyte_cdk.sources.streams.http import HttpStream
from source_webflow.auth import WebflowTokenAuthenticator
class MyWebflowStream(HttpStream):
def __init__(self, authenticator: WebflowTokenAuthenticator, **kwargs):
super().__init__(authenticator=authenticator, **kwargs)
# The authenticator will automatically be used for all HTTP requests
# made by this streamInstall with Tessl CLI
npx tessl i tessl/pypi-airbyte-source-webflow