upbeatBytes/goodnews/api.py

"""FastAPI service for goodNews.

A read-only JSON API over the ingestion database, plus a small static site that
consumes it. The same endpoints back both the website and any future companion
app; the auto-generated OpenAPI docs at /docs are that shared contract.

Run with the bundled CLI:  goodnews serve
Or directly:               uvicorn goodnews.api:app --host 0.0.0.0 --port 8000

The database path comes from GOODNEWS_DB (falling back to the repo's data dir),
so the API and CLI always read the same file.
"""

from __future__ import annotations

import os
import sqlite3
from contextlib import contextmanager
from pathlib import Path

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from . import queries
from .db import connect, init_db
from .taxonomy import FLAVORS, TOPICS

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_DB = ROOT / "data" / "goodnews.sqlite3"
STATIC_DIR = Path(__file__).resolve().parent / "static"


def db_path() -> Path:
    return Path(os.environ.get("GOODNEWS_DB", str(DEFAULT_DB)))


@contextmanager
def get_conn():
    conn = connect(db_path())
    try:
        yield conn
    finally:
        conn.close()


# --- Response models (the companion-app contract) ---------------------------


class Category(BaseModel):
    key: str
    description: str


class CategoriesResponse(BaseModel):
    topics: list[Category]
    flavors: list[Category]


class CategoryCount(BaseModel):
    topic: str | None
    flavor: str | None
    count: int


class Article(BaseModel):
    id: int
    title: str
    description: str | None = None
    url: str
    image_url: str | None = None
    published_at: str | None = None
    source: str
    topic: str | None = None
    flavor: str | None = None
    accepted: bool
    rank_score: int | None = None
    reason_code: str | None = None
    reason_text: str | None = None
    model_name: str | None = None
    rank: int | None = None  # position within a brief, when applicable

    @classmethod
    def from_row(cls, row: dict) -> "Article":
        return cls(
            id=row["id"],
            title=row["title"],
            description=row.get("description"),
            url=row["canonical_url"],
            image_url=row.get("image_url"),
            published_at=row.get("published_at"),
            source=row["source_name"],
            topic=row.get("topic"),
            flavor=row.get("flavor"),
            accepted=bool(row.get("accepted")),
            rank_score=row.get("rank_score"),
            reason_code=row.get("reason_code"),
            reason_text=row.get("reason_text"),
            model_name=row.get("model_name"),
            rank=row.get("rank"),
        )


class FeedResponse(BaseModel):
    topic: str | None
    flavor: str | None
    count: int
    items: list[Article]


class BriefResponse(BaseModel):
    brief_date: str | None
    title: str | None
    items: list[Article]


# --- App --------------------------------------------------------------------


def create_app() -> FastAPI:
    app = FastAPI(
        title="goodNews API",
        version="0.1.0",
        description="Constructive, uplifting news — metadata and links only.",
    )

    # The website and companion app may live on other origins; allow them.
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_methods=["GET"],
        allow_headers=["*"],
    )

    @app.get("/healthz")
    def healthz() -> dict:
        with get_conn() as conn:
            init_db(conn)
            scored = conn.execute("SELECT COUNT(*) FROM article_scores").fetchone()[0]
        return {"status": "ok", "scored_articles": scored}

    @app.get("/api/categories", response_model=CategoriesResponse)
    def categories() -> CategoriesResponse:
        return CategoriesResponse(
            topics=[Category(key=k, description=v) for k, v in TOPICS.items()],
            flavors=[Category(key=k, description=v) for k, v in FLAVORS.items()],
        )

    @app.get("/api/category-counts", response_model=list[CategoryCount])
    def category_counts(accepted_only: bool = True) -> list[CategoryCount]:
        with get_conn() as conn:
            rows = queries.category_counts(conn, accepted_only=accepted_only)
        return [CategoryCount(**row) for row in rows]

    @app.get("/api/feed", response_model=FeedResponse)
    def feed(
        topic: str | None = Query(None),
        flavor: str | None = Query(None),
        accepted_only: bool = True,
        limit: int = Query(30, ge=1, le=100),
        offset: int = Query(0, ge=0),
    ) -> FeedResponse:
        if topic and topic.lower() not in TOPICS:
            raise HTTPException(400, f"unknown topic: {topic}")
        if flavor and flavor.lower() not in FLAVORS:
            raise HTTPException(400, f"unknown flavor: {flavor}")
        with get_conn() as conn:
            rows = queries.feed(
                conn,
                topic=topic,
                flavor=flavor,
                accepted_only=accepted_only,
                limit=limit,
                offset=offset,
            )
        return FeedResponse(
            topic=topic,
            flavor=flavor,
            count=len(rows),
            items=[Article.from_row(r) for r in rows],
        )

    @app.get("/api/brief", response_model=BriefResponse)
    def brief(date: str | None = Query(None), limit: int = Query(10, ge=1, le=50)) -> BriefResponse:
        with get_conn() as conn:
            data = queries.brief(conn, brief_date=date, limit=limit)
        return BriefResponse(
            brief_date=data["brief_date"],
            title=data["title"],
            items=[Article.from_row(r) for r in data["items"]],
        )

    @app.get("/api/brief-dates", response_model=list[str])
    def brief_dates(limit: int = Query(30, ge=1, le=365)) -> list[str]:
        with get_conn() as conn:
            return queries.available_dates(conn, limit=limit)

    # Static site last, mounted at root, so /api/* and /healthz win.
    if STATIC_DIR.is_dir():
        app.mount("/", StaticFiles(directory=str(STATIC_DIR), html=True), name="site")

    return app


app = create_app()