rss_feed_extract:
  type: kedro_workbench.extras.datasets.RSSDataSet.RSSFeedExtract
  url: <https://api.msrc.microsoft.com/update-guide/rss>

rss_feed_load:
  type: kedro_workbench.extras.datasets.RSSDataSet.RSSFeedLoad
  mongo_url: "mongodb+srv://<username>:<password>@bighatcluster.wamzrdr.mongodb.net/"
  mongo_db: "TBD"
  mongo_collection: "TBD"
  mongo_table: "TBD"
  credentials: mongo_atlas

def extract_rss_feed() -> Dict[str, Any]:
    raw_rss_feed = RSSFeedExtract() # Q. how does the catalog 'url' value get passed to the __init__ method?
    raw_rss_feed.load()
    
    return {'key_1':'value_1', 'key_2': 'value_2'}
    
    
def transform_rss_feed(raw_rss_feed: Dict[str, Any]) -> List[Dict[str, Any]]:
    
    return [{'key_1_T':'value_1_T', 'key_2_T': 'value_2_T'}]
    
    
def load_rss_feed(prepped_rss_items: List[Dict[str, Any]]) -> None:
    rss_feed_load = RSSFeedLoad(prepped_rss_items) # not clear how to create the custom dataset that takes data from catalog and credentials and the previous node
    rss_feed_load.save()

pipeline([
        node(
            func=extract_rss_feed,
                inputs=None,
                outputs='rss_feed_for_transforming',
                name="extract_rss_feed",
        ),
        node(
            func=transform_rss_feed,
                inputs="rss_feed_for_transforming",
                outputs='rss_for_loading',
                name="transform_rss_items",
        ),
        node(
            func=load_rss_feed,
                inputs="rss_for_loading",
                outputs="rss_feed_load",
                name="load_rss_items",
        ),

    ])

class RSSFeedExtract(AbstractDataSet):
    def __init__(self, url: str):
        self._url = URL

class RSSFeedLoad(AbstractDataSet):
    def __init__(self, mongo_url: str, mongo_db: str, mongo_collection: str, mongo_table: str, credentials: Dict[str, Any], data: Any = None):
        self._data = data # comes from the previous node
        self._mongo_url = mongo_url
        self._mongo_db = mongo_db
        self._mongo_collection = mongo_collection
        self._mongo_table = mongo_table
        self._username = credentials['username']
        self._password = credentials['password']

type

__init__

class RSSFeedExtract(url, load_args=None, save_args=None):
    ...

__init__

I created a basic API Dataset that specified a url parameter and I didn't need to pass the URL into the node function.

url

RSSFeedExtract._load

_load

__init__

1. how do I set up the node function/pipeline configure for a dataset that is readonly, ie, fetching an RSS feed

_save

APIDataSet

from kedro_workbench.extras.datasets.RSSDataSet import RSSFeedExtract, RSSFeedLoad
def extract_rss_feed() -> Dict[str, Any]:
    raw_rss_feed = RSSFeedExtract() 
    raw_rss_feed.load()
    
    return {'key_1':'value_1', 'key_2': 'value_2'}

node(
            func=extract_rss_feed,
                inputs=None,
                outputs='rss_feed_for_transforming',
                name="extract_rss_feed",
        ),