tessl/pypi-airbyte-source-webflow
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
source-configuration.mddocs/
Source Configuration
Main source connector functionality for configuration, connection validation, and stream management. The SourceWebflow class implements Airbyte's AbstractSource interface and provides the core connector behavior.
Capabilities
Source Class
The main connector class that handles configuration, authentication, connection validation, and stream discovery.
class SourceWebflow(AbstractSource):
"""Main source class for Webflow connector."""Connection Validation
Validates that the provided configuration can successfully connect to the Webflow API by attempting to read collection metadata.
def check_connection(self, logger: logging.Logger, config: Mapping[str, Any]) -> Tuple[bool, any]:
"""
Validate connection to Webflow API.
Parameters:
- logger: Logger instance for connection status messages
- config: Configuration dictionary with api_key, site_id, and optional accept_version
Returns:
Tuple of (success_boolean, error_or_none)
"""Stream Discovery
Discovers available Webflow collections and generates corresponding stream objects for data extraction.
def streams(self, config: Mapping[str, Any]) -> List[Stream]:
"""
Generate list of available streams based on Webflow collections.
Parameters:
- config: Configuration dictionary with authentication and site information
Returns:
List of Stream objects, one for each discovered Webflow collection
"""Stream Generation
Internal method that creates stream objects for each discovered collection.
def generate_streams(self, authenticator: WebflowTokenAuthenticator, site_id: str) -> List[Stream]:
"""
Generate stream objects for each collection.
Parameters:
- authenticator: Configured Webflow authenticator instance
- site_id: Webflow site identifier
Returns:
Generator yielding CollectionContents stream objects
"""Authentication Factory
Static method to create properly configured authenticator instances from configuration.
@staticmethod
def get_authenticator(config):
"""
Create WebflowTokenAuthenticator from configuration.
Parameters:
- config: Configuration dictionary containing api_key and optional accept_version
Returns:
WebflowTokenAuthenticator instance
Raises:
Exception: If api_key is missing from configuration
"""Collection Name Mapping
Internal utility to map collection names to their corresponding IDs for API calls.
@staticmethod
def _get_collection_name_to_id_dict(authenticator = None, site_id: str = None) -> Mapping[str, str]:
"""
Create mapping from collection names to collection IDs.
Parameters:
- authenticator: Webflow authenticator instance
- site_id: Webflow site identifier
Returns:
Dictionary mapping collection names to their API IDs
"""Usage Examples
Basic Connection Test
from source_webflow import SourceWebflow
import logging
config = {
"api_key": "your_api_token",
"site_id": "your_site_id"
}
source = SourceWebflow()
logger = logging.getLogger()
success, error = source.check_connection(logger, config)
if success:
print("Connected successfully!")
else:
print(f"Connection failed: {error}")Discovering Available Streams
from source_webflow import SourceWebflow
config = {
"api_key": "your_api_token",
"site_id": "your_site_id"
}
source = SourceWebflow()
streams = source.streams(config)
for stream in streams:
print(f"Collection: {stream.name}")
print(f"Schema: {stream.get_json_schema()}")Configuration Requirements
The configuration dictionary must contain:
api_key(string, required): Webflow API token for authenticationsite_id(string, required): Identifier of the Webflow site to extract data fromaccept_version(string, optional): API version header, defaults to "1.0.0"
Error Handling
Connection validation will return (False, error) for common issues:
- Invalid or missing API key
- Inaccessible site ID
- Network connectivity problems
- Webflow API errors or rate limiting
The error object contains details about the specific failure for debugging purposes.
Install with Tessl CLI
npx tessl i tessl/pypi-airbyte-source-webflow