Skip to content

hackernews

hackernews

Hacker News connector — top stories from the HN Firebase API.

No authentication required. All API calls are in module-level functions for easy mocking in tests.

Classes

HackerNewsConnector 

HackerNewsConnector()

Bases: BaseConnector

Fetch the current top stories from Hacker News.

Source code in src/diapason/connectors/hackernews.py 
42
43
def __init__(self) -> None:
    self._status = SyncStatus()
Methods:
sync 
sync(
    *,
    since: Optional[datetime] = None,
    cursor: Optional[str] = None,
) -> Iterator[Document]

Yield Documents for the top 5 Hacker News stories.

Source code in src/diapason/connectors/hackernews.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def sync(
    self, *, since: Optional[datetime] = None, cursor: Optional[str] = None
) -> Iterator[Document]:
    """Yield Documents for the top 5 Hacker News stories."""
    top_ids = _hn_top_story_ids()

    for story_id in top_ids[:5]:
        item = _hn_item(story_id)
        if item is None:
            continue

        title = item.get("title", "")
        score = item.get("score", 0)
        descendants = item.get("descendants", 0)
        url = item.get("url", "")
        by = item.get("by", "")
        ts = datetime.now()
        if item.get("time"):
            ts = datetime.fromtimestamp(item["time"])

        yield Document(
            doc_id=f"hn-{story_id}",
            source="hackernews",
            doc_type="story",
            content=f"Score: {score}, Comments: {descendants}",
            title=title,
            author=by,
            timestamp=ts,
            url=url or None,
            metadata={
                "story_id": story_id,
                "score": score,
                "descendants": descendants,
            },
        )

    self._status.state = "idle"
    self._status.last_sync = datetime.now()