65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
import asyncio
|
|
from datetime import datetime
|
|
|
|
import dateutil.parser
|
|
import httpx
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class Story(BaseModel):
|
|
"""Model representing a Hacker News story."""
|
|
|
|
id: str
|
|
title: str = Field(description="Title of the story.")
|
|
url: str | None = Field(description="URL of the story.")
|
|
|
|
created_at: datetime
|
|
|
|
|
|
class HackerNewsClient:
|
|
def __init__(self, client: httpx.AsyncClient | None = None):
|
|
base_url = "https://hn.algolia.com/api/v1"
|
|
if client:
|
|
client.base_url = base_url
|
|
self._client = client
|
|
else:
|
|
self._client = httpx.AsyncClient(base_url=base_url)
|
|
|
|
async def get_top_stories(self, limit: int = 10) -> list[Story]:
|
|
resp = await self._client.get(
|
|
"search",
|
|
params={"tags": "front_page", "hitsPerPage": limit, "page": 0},
|
|
)
|
|
resp.raise_for_status()
|
|
return [
|
|
Story(
|
|
id=hit["objectID"],
|
|
title=hit["title"],
|
|
url=hit.get("url"),
|
|
created_at=dateutil.parser.isoparse(hit["created_at"]),
|
|
)
|
|
for hit in resp.json().get("hits", [])
|
|
]
|
|
|
|
async def get_item(self, item_id):
|
|
resp = await self._client.get(f"items/{item_id}")
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def close(self):
|
|
"""Close the underlying HTTP client."""
|
|
|
|
if self._client and not self._client.is_closed:
|
|
await self._client.aclose()
|
|
|
|
def __del__(self):
|
|
"""Ensure the HTTP client is closed when the object is deleted."""
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_running():
|
|
loop.create_task(self.close())
|
|
else:
|
|
loop.run_until_complete(self.close())
|
|
except Exception:
|
|
pass
|