Airbyte Source RSS

from source_rss import SourceRss
from source_rss.run import run
from source_rss.components import CustomExtractor

from airbyte_cdk.entrypoint import launch

from source_rss import SourceRss
from airbyte_cdk.entrypoint import launch
import sys

# Create and launch the source connector
source = SourceRss()
launch(source, sys.argv[1:])

# Get connector specification (returns JSON schema for configuration)
poetry run source-rss spec

# Test connection with config (validates RSS URL accessibility)
poetry run source-rss check --config config.json

# Discover available streams (returns items stream schema)
poetry run source-rss discover --config config.json

# Extract data (reads RSS feed items according to catalog configuration)
poetry run source-rss read --config config.json --catalog catalog.json

# Extract with state for incremental sync
poetry run source-rss read --config config.json --catalog catalog.json --state state.json

{
  "url": "https://example.com/rss.xml"
}

class SourceRss(YamlDeclarativeSource):
    """
    Declarative source connector for RSS feeds.
    
    Inherits from YamlDeclarativeSource and loads configuration from manifest.yaml.
    """
    
    def __init__(self):
        """Initialize SourceRss with manifest.yaml configuration."""

class CustomExtractor(RecordExtractor):
    """
    Custom RSS feed parser and record extractor.
    
    Processes RSS feed XML responses and extracts structured item data
    with timestamp-based filtering for incremental synchronization.
    Uses feedparser library for robust RSS/Atom parsing and pytz for timezone handling.
    """
    
    def extract_records(self, response: requests.Response, **kwargs) -> List[Mapping[str, Any]]:
        """
        Extract and transform RSS feed items from HTTP response.
        
        Parses RSS/Atom feed XML using feedparser, converts items to structured records,
        applies timestamp filtering for incremental sync, and handles timezone conversions.
        
        Args:
            response (requests.Response): HTTP response containing RSS feed XML
            **kwargs: Additional extraction parameters
            
        Returns:
            List[Mapping[str, Any]]: List of extracted RSS items as dictionaries
            
        Extracted Fields:
            - title (str, optional): RSS item title
            - link (str, optional): RSS item URL
            - description (str, optional): RSS item description/content  
            - author (str, optional): RSS item author
            - category (str, optional): RSS item category
            - comments (str, optional): RSS item comments URL
            - enclosure (str, optional): RSS item enclosure/attachment
            - guid (str, optional): RSS item unique identifier
            - published (str): RSS item publication date in ISO format with UTC timezone
            
        Processing:
            - Extracts items from feed.entries in reverse order (oldest first)
            - Converts published_parsed timestamps to UTC ISO format
            - Filters items based on feed-level publication date for incremental sync
            - Handles missing fields gracefully (sets to null)
        """

def run():
    """
    Create SourceRss instance and launch connector via airbyte_cdk.entrypoint.launch.
    
    Uses sys.argv[1:] for command line arguments processing.
    Supports standard Airbyte connector commands: spec, check, discover, read.
    """

# RSS Items Stream Schema
{
    "type": "object",
    "additionalProperties": True,
    "required": ["published"],
    "properties": {
        "title": {"type": ["null", "string"]},
        "link": {"type": ["null", "string"]}, 
        "description": {"type": ["null", "string"]},
        "author": {"type": ["null", "string"]},
        "category": {"type": ["null", "string"]},
        "comments": {"type": ["null", "string"]},
        "enclosure": {"type": ["null", "string"]},
        "guid": {"type": ["null", "string"]},
        "published": {"type": "string", "format": "date-time"}
    }
}

import pytest

# Connector acceptance test fixture
@pytest.fixture(scope="session", autouse=True)
def connector_setup():
    """
    Placeholder fixture for external resources that acceptance test might require.
    """

from source_rss import SourceRss
from airbyte_cdk.entrypoint import launch

# Initialize connector
source = SourceRss()

# Configuration for RSS feed
config = {
    "url": "https://www.nasa.gov/rss/dyn/breaking_news.rss"
}

# The connector will extract RSS items with fields:
# - title, link, description
# - author, category, comments  
# - enclosure, guid
# - published (ISO datetime)

from source_rss.components import CustomExtractor
import requests

# Create custom extractor instance
extractor = CustomExtractor()

# Process RSS feed response
response = requests.get("https://example.com/feed.rss")
records = extractor.extract_records(response)

# Each record contains RSS item fields
for record in records:
    print(f"Title: {record.get('title')}")
    print(f"Published: {record.get('published')}")
    print(f"Link: {record.get('link')}")

# Build connector image
airbyte-ci connectors --name=source-rss build

# Run connector commands
docker run --rm airbyte/source-rss:dev spec
docker run --rm -v $(pwd)/config:/config airbyte/source-rss:dev check --config /config/config.json
docker run --rm -v $(pwd)/config:/config airbyte/source-rss:dev discover --config /config/config.json
docker run --rm -v $(pwd)/config:/config -v $(pwd)/catalog:/catalog airbyte/source-rss:dev read --config /config/config.json --catalog /catalog/catalog.json

# Core Airbyte framework
airbyte-cdk = "^0"

# RSS/Atom feed parsing
feedparser = "6.0.10" 

# Timezone handling
pytz = "2022.6"

# Testing framework
pytest = "*"
pytest-mock = "*"

# HTTP mocking for tests  
requests-mock = "*"