dagster_software_defined_assets_hacker_news_topstories.py

1import json
2import os
3import requests
4from dagster import asset, Definitions
5
6@asset
7def topstory_ids():
8    url = "https://hacker-news.firebaseio.com/v0/topstories.json"
9    content = requests.get(url).json()
10    with open("topstory_ids.json", "w") as f:
11        json.dump(content[:10], f)
12
13@asset(deps=[topstory_ids])
14def topstories():
15    with open("topstory_ids.json", "r") as f:
16        topstory_ids = json.load(f)
17
18    results = []
19    for item_id in topstory_ids:
20        item = requests.get(f"https://hacker-news.firebaseio.com/v0/item/{item_id}.json").json()
21        results.append(item)
22
23    with open("topstories.json", "w") as f:
24        json.dump(results, f)
25
26defs = Definitions(
27    assets=[topstory_ids, topstories],
28)