Build It: A Daily Data Pipeline With LLM Enrichment

# pyproject.toml: httpx, pandas, anthropic, sqlalchemy import asyncio import httpx import pandas as pd from sqlalchemy import create_engine, text from anthropic import AsyncAnthropic DB_URL = "sqlite:///pipeline.db" engine = create_engine(DB_URL) def init_db(): with engine.begin() as conn: conn.execute(text(""" CREATE TABLE IF NOT EXISTS stories ( id INTEGER PRIMARY KEY, title TEXT NOT NULL, url TEXT, fetched_at TEXT NOT NULL, sentiment TEXT, UNIQUE(id) ) """))

async def fetch_stories() -> pd.DataFrame: url = "https://hacker-news.firebaseio.com/v0/topstories.json" async with httpx.AsyncClient(timeout=10) as client: ids = (await client.get(url)).json()[:20] async def get_one(sid): r = await client.get(f"https://hacker-news.firebaseio.com/v0/item/{sid}.json") return r.json() raw = await asyncio.gather(*(get_one(i) for i in ids)) df = pd.DataFrame(raw)[["id", "title", "url"]].dropna(subset=["title"]) df["fetched_at"] = pd.Timestamp.utcnow().isoformat() return df

client = AsyncAnthropic() sem = asyncio.Semaphore(5) async def classify(title: str) -> str: async with sem: try: r = await client.messages.create( model="claude-haiku-4-5", # cheaper for bulk classification max_tokens=10, messages=[{ "role": "user", "content": f"Classify sentiment of this headline as exactly one word: positive, negative, or neutral. No other text.\n\nHeadline: {title}" }], ) word = r.content[0].text.strip().lower() return word if word in {"positive", "negative", "neutral"} else "neutral" except Exception as e: print(f"classify failed: {e}") return "unknown" async def enrich(df: pd.DataFrame) -> pd.DataFrame: sentiments = await asyncio.gather(*(classify(t) for t in df["title"])) df = df.copy() df["sentiment"] = sentiments return df

def upsert(df: pd.DataFrame) -> int: sql = text(""" INSERT INTO stories (id, title, url, fetched_at, sentiment) VALUES (:id, :title, :url, :fetched_at, :sentiment) ON CONFLICT(id) DO UPDATE SET sentiment = excluded.sentiment, fetched_at = excluded.fetched_at """) rows = df.to_dict(orient="records") with engine.begin() as conn: conn.execute(sql, rows) return len(rows) async def main(): init_db() df = await fetch_stories() df = await enrich(df) n = upsert(df) print(f"Inserted/updated {n} rows.") # Read back top negatives with engine.connect() as conn: neg = pd.read_sql("SELECT title FROM stories WHERE sentiment = 'negative' LIMIT 5", conn) print("\nRecent negative headlines:") print(neg.to_string(index=False)) asyncio.run(main())