👋 I am just starting with kedro. I am putting together an example pipeline to make sure I understand all the concepts before I build a larger project with it. One concept I have not able to fully figure out is how to work with APIs. The simple project I am trying to build is: 1. get top N articles on hacker news 2. fetch the items, to get the urls 3. use an api to summarize the url 4. save the summary, url, etc I am able to easily build the HackerNewsTopAPIDataset for getting the top items. However, I am not able to figure out how to get those item ids into the HackerNewsItemsAPIDataset. I am of course able to fill up the node function with all kinds of io and get it to work, which is what I did. However, everything I read says this is the wrong approach and node functions should be purely functional. I have stumbled into the stackoverflow question, https://stackoverflow.com/questions/73430557/dynamic-parameters-on-datasets-in-kedro, which talks about adding kwargs to the _load(). However I do not see how to pass arguments into the load without explicitly pulling datasets into the node function and calling it directly. This brings me back to doing io in node functions. Now I am left scratching my head on how to link datasets that require input to be able to function. Any insights or pointers would be greatly appreciated.

What you’re running into is a pattern with dynamic inputs - APIs where the second call depends on the result of the first. Kedro datasets are designed to be declarative and static (configured in the catalog), so they don’t directly support passing runtime values into

_load()

. That’s why it feels awkward. The recommended approach in Kedro is: • Keep API I/O inside datasets, not in node functions. • If you need runtime parameters (like the list of HackerNews IDs), capture them as data flowing through the pipeline, not as dataset arguments. So in your example: 1. Create one dataset (

HackerNewsTopAPIDataset

) that loads the top N IDs. 2. Write a node function that takes those IDs and transforms them into a format that your next dataset can understand (for example, a list of IDs or a table). 3. Create another dataset (

HackerNewsItemsAPIDataset

) that is parametrized to fetch items for a given list of IDs - but instead of trying to pass the IDs directly into the dataset config, you make the dataset read from an intermediate file or memory dataset produced by the previous node. In other words, rather than trying to call

.load()

on a dataset inside your node, you let Kedro orchestrate: • Node 1 produces the IDs written to a dataset (

MemoryDataset

or

JSONDataSet

) • Node 2 consumes that dataset (the IDs), and calls your items API • Node 3 consumes the items and calls the summarization API • Node 4 saves the outputs Alternatively if you really need runtime kwargs: Kedro does have a hook (

before_dataset_loaded

) where you can inject arguments dynamically at runtime (like changing the

url

or

params

of an API dataset).

Oh! The mistake in my thinking was having a dataset equate to a single datasource. And assuming I should only pass information through the pipeline/node functions. When a dataset class can pull in N other datasets to create a single view! This was very helpful, thank you!

Datasets should not depend on the catalog. They’re meant to be small, declarative wrappers around I/O (load/save a file, call an API, etc.), configured in the catalog. If a dataset reaches into the catalog to load another dataset, you end up with hidden dependencies and break reproducibility. That’s why you won’t find an official

catalog

object in datasets.

I think the thing I am missing is how HackerNewsItemsDataset gets the json file. If its not meant to be available through the catalog and the ids should not be passed in via a function in the node, then I am at a loss how to refer to the id dataset.

Your use case is a bit tricky because you want to configure the dataset dynamically, though normally, it’s static and done via catalog. That’s why I suggest a workaround here.

# catalog.yml

hn_ids_file:
  type: kedro.extras.datasets.json.JSONDataSet
  filepath: data/01_raw/hn_ids.json

hn_items:
  type: path.to.HackerNewsItemsDataset
  ids_filepath: data/01_raw/hn_ids.json
  ...

# nodes.py
def identity_save_ids(ids):
    # simply return ids and Kedro will save it to hn_ids_file if the node outputs are configured accordingly
    return hn_ids_file

# pipeline.py

from kedro.pipeline import Pipeline, node, pipeline
from .nodes import identity_save_ids

def create_pipeline(**kwargs) -> Pipeline:
    return pipeline(
        [
            # 1) load top ids, 2) save them to JSON dataset (hn_ids_file),
            node(identity_save_ids, inputs="hn_top", outputs="hn_ids_file", name="save_ids"),
            # 3) when some node needs hn_items, Kedro will call hn_items.load(),
            #    which reads ids_filepath and fetches articles
        ]
    )

This is what I came up with:

from typing import Optional
import aiohttp
import asyncio
import pandas as pd
from <http://kedro.io|kedro.io> import AbstractDataset
from <http://kedro.io|kedro.io> import DataCatalog

# Uses api defined at <https://github.com/HackerNews/API?tab=readme-ov-file>
class HackerNewsUrlsDataset(AbstractDataset):
    # dependencies is a list of other datasets defined in the catalog
    def __init__(self, url: str, dependencies: list[str]):
        self._url = url
        self._catalog: Optional[DataCatalog] = None
        self._dependencies = dependencies
        self._datasets = {}

    def _describe(self):
        return {
            "url": self._url
        }

    def inject_catalog(self, catalog):
        self._catalog = catalog

    def load(self):
        if not self._catalog:
            raise RuntimeError("catalog has not been injected into HackerNewsUrlsDataset")

        datasets = self._load_datasets()

        return self.get_items(datasets['hn_top'])

    def save(self, data: pd.DataFrame):
        raise NotImplementedError("HackerNewsUrls is a readonly api")


    def _load_datasets(self):
        datasets = {}
        for dependency in self._dependencies:
            datasets[dependency] = self._catalog.load(dependency)
        return datasets


    def get_items(self, item_ids: pd.DataFrame) -> pd.DataFrame:
        # <https://hacker-news.firebaseio.com/v0/item/{id}.json> -> <https://hacker-news.firebaseio.com/v0/item/1234567.json>
        urls = [self._url.format(id=item_id["item_id"]) for _, item_id in item_ids.iterrows()]

        items_df = asyncio.run(self._fetch_all(urls))
        return items_df

    async def _fetch_all(self, urls: list[str]) -> pd.DataFrame:
        async with aiohttp.ClientSession() as session:
            tasks = [self._fetch_url(session, url) for url in urls]
            results = await asyncio.gather(*tasks)
            return pd.DataFrame(results)

    async def _fetch_url(self, session: aiohttp.ClientSession, url: str) -> dict:
        async with session.get(url) as response:
            return await response.json()