How to connect csv, api (post) and json dataset?

[afuetterer]

Hi,

can anybody help me with the connection of a pandas.CSVDataSet, a api.APIDataSet and a json.JSONDataSet? I want to iterate over the entries in the csv file and one by one use them in a post request to the api and then collect the responses in a single json file.

A toy example: Let's say I have a csv file called friends.csv containing

name
Joey
Chandler
Ross
Phoebe
Rachel
Monica

in my data catalog

friends_csv:
  type: pandas.CSVDataSet
  filepath: data/01_raw/friends.csv

users_api_post:
  type: api.APIDataSet
  url: https://reqres.in/api/users
  method: POST

friends_json:
  type: json.JSONDataSet
  filepath: data/02_intermediate/friends.json

I can access the api programmatically.

from kedro.extras.datasets.api import APIDataSet

api_results = []
for friend_name in friends["name"]:
    response = APIDataSet(
        url="https://reqres.in/api/users",
        method="POST",
        json={"first_name": friend_name}
    ).load()
    api_results.append(response.json())
print(api_results)
# [{'first_name': 'Joey', 'id': '480', 'createdAt': '2022-05-11T13:44:58.714Z'}, {'first_name': 'Chandler', 'id': '894', 'createdAt': '2022-05-11T13:44:58.962Z'}, {'first_name': 'Ross', 'id': '905', 'createdAt': '2022-05-11T13:44:59.180Z'}, {'first_name': 'Phoebe', 'id': '799', 'createdAt': '2022-05-11T13:44:59.355Z'}, {'first_name': 'Rachel', 'id': '576', 'createdAt': '2022-05-11T13:44:59.532Z'}, {'first_name': 'Monica', 'id': '333', 'createdAt': '2022-05-11T13:44:59.715Z'}]

catalog.save("friends_json", api_results)

Can this be done more elegantly, e.g. using the dataset definition from the data catalog? I am unsure how to feed the arguments to the api dataset, which are the contents of the csv dataset.
Where would you do the iteration part, e.g. how many nodes would you use to connect these three datasets?

Any feedback is greatly appreciated.

[antonymilne]

This is a very interesting question! Thanks for explaining it all so clearly.

Unfortunately the default APIDataSet isn't really well suited to this because it implements the load method but not a save one. What you're doing is quite unusual because no one has ever asked for us to implement a save method for it! The separation between I/O (which belongs in the catalog) and data processing (which belongs in a node) isn't so well defined in your example so there's not really one correct way of doing it.

The simplest solution is just to not use APIDataSet at all and to instead perform the request inside a node:

def users_api_post(friends: pd.CSVDataSet) -> List[Dict[str, Any]]:
    api_results = []
    for friend_name in friends["name"]:
        # Taken from the APIDataSet source code, which also does some error handling and a few other bits
        response = requests.request(
            url="https://reqres.in/api/users",
            method="POST",
            json={"first_name": friend_name}
        )
    api_results.append(response.json())
    return api_results

# node(users_api_post, friends_csv, friends_json)

This is a bit naughty because your node is no longer really a pure function, but you could also regard that request as doing some kind of data transformation maybe 🤔 It's not a terrible thing to do.

If you really want to use a dataset for it then you'll need to write your own custom dataset that somehow combines APIDataSet and JSONDataSet. It's still possible to do this with one node:

class MyAPIDataSet(APIDataSet):
    def __init__(self, *args, **kwargs):
        self._filepath = kwargs.pop("filepath")
        super().__init__(*args, **kwargs)
    
    def _save(self, data: Any) -> None:
        api_results = []
        for friend_name in data:
            response = requests.request(
                url="https://reqres.in/api/users",
                method="POST",
                json={"first_name": friend_name}
            )
        api_results.append(response.json())

        # I just handle the simple local file case here. You could also do the full case in JSONDataSet
        # which also handles credentials, remote file storage, etc.
        json.dump(api_results, filepath)

def extract_names(friends: pd.DataFrame) -> pd.Series:
     return friends["name"]

# node(extract_names, friends_csv, friends_api_json)

In the catalog you'd do:

friends_api_json:
  type: path.to.MyAPIDataSet
  url: https://reqres.in/api/users
  method: POST
  filepath: data/02_intermediate/friends.json

This is also a bit weird but not a terrible thing to do.

[afuetterer]

Thanks for your reply @AntonyMilneQB!

I have only recently started working with kedro. It is really a great tool.

Of course the example is not realistic, but it illustrates my use case. Some API requests need a single id or query as their parameter. For example, you have a list of IDs and want to send them one by one to an API.

I also considered the variant of describing the logic in a node, but I thought there had to be a better way. Thanks for your two suggestions for my edge case. I will consider them both and let know you which way I went.

[antonymilne]

Cool, thanks for the comments! Your use case definitely makes sense, and as it stands we're definitely not set up to handle it well natively in kedro.

Possibly the APIDataSet should be made more generic (or maybe a new dataset added) to accommodate this sort of thing. Ideally I thought we could use PartionedDataSet for this, which is used for a similar purpose to wrap an iterator over multiple instances of a dataset (like saving many csv files to one directory). Unfortunately our current implementation of that is very heavily tied to file-based partitioning though so can't be so easily extended to iterate over APIDataSet. Ultimately the cleanest/most kedroic solution would probably be to have some generic IteratorDataSet which can wrap any dataset, including APIDataSet. But that's a lot of work 😅 I'll be interested in hearing if anyone else is trying to solve this since I think you're the first person who has brought it up.

[afuetterer]

Initially, I thought about using PartionedDataSet for this. I thought I could store each API JSON response as a single JSON file and load those with a PartionedDataSet. But if the number of responses exceeds a certain number (which I cannot know in advance), this approach seemed impractical to me.

An IteratorDataSet sounds like a nice idea, but of course it should be needed/used by the community.

Thanks again for your feedback on this issue!