main.py

import requests
import pandas as pd
from prefect import task, Flow
import json



@task
def extract(url_from):
    response = requests.get(url_from)
    if response:
        return json.loads(response.content)["results"]
    else:
        print("No response available.")



@task
def transform(data_dict):
    people_list = []
    for person in data_dict:
        single_item = {
            'gender': person["gender"],
            "name": person["name"]["title"] + person["name"]["first"] + person["name"]["last"],
            "nat": "AU",
        }
        people_list.append(single_item)
    
    # return dataframe from list of dicts
    return pd.DataFrame(people_list)



@task
def load(data_df, filename):
    data_df.to_csv(f"{filename}.csv", index=False)



def start_data_collection(num_people_to_fetch):
    with Flow("Random User API ETL:") as flow:
        # get 7 people profile upon each request
        people = extract(f'https://randomuser.me/api/?inc=gender,name,nat&results={num_people_to_fetch}')
        # make a dataframe out of the response
        user_df = transform(people)
        # save the dataframe formed to disk
        load(user_df, f'{num_people_to_fetch}_people')
    return flow



if __name__ == "__main__":
    flow = start_data_collection(3)
    flow.run()