From 2e22a2b3ff25f21f8afb07809be2a489adb6c037 Mon Sep 17 00:00:00 2001 From: cottongin Date: Thu, 12 Mar 2026 01:40:04 -0400 Subject: [PATCH] feat: add FastAPI routes for playlist, shows, admin, and health Made-with: Cursor --- src/ntr_fetcher/api.py | 130 +++++++++++++++++++++++++++++++++++++++++ tests/test_api.py | 119 +++++++++++++++++++++++++++++++++++++ 2 files changed, 249 insertions(+) create mode 100644 src/ntr_fetcher/api.py create mode 100644 tests/test_api.py diff --git a/src/ntr_fetcher/api.py b/src/ntr_fetcher/api.py new file mode 100644 index 0000000..3140410 --- /dev/null +++ b/src/ntr_fetcher/api.py @@ -0,0 +1,130 @@ +import logging +from datetime import datetime, timezone + +from fastapi import FastAPI, HTTPException, Depends, Header +from pydantic import BaseModel + +from ntr_fetcher.db import Database +from ntr_fetcher.week import get_show_week + +logger = logging.getLogger(__name__) + + +class RefreshRequest(BaseModel): + full: bool = False + + +class AddTrackRequest(BaseModel): + soundcloud_url: str | None = None + track_id: int | None = None + position: int | None = None + + +class MoveTrackRequest(BaseModel): + position: int + + +def create_app(db: Database, poller, admin_token: str) -> FastAPI: + app = FastAPI(title="NtR SoundCloud Fetcher") + + def _require_admin(authorization: str | None = Header(None)): + if authorization is None or not authorization.startswith("Bearer "): + raise HTTPException(status_code=401, detail="Missing or invalid token") + token = authorization.removeprefix("Bearer ") + if token != admin_token: + raise HTTPException(status_code=401, detail="Invalid token") + + def _current_show(): + now = datetime.now(timezone.utc) + week_start, week_end = get_show_week(now, show_day=2, show_hour=22) + return db.get_or_create_show(week_start, week_end) + + @app.get("/health") + def health(): + show = _current_show() + tracks = db.get_show_tracks(show.id) + return { + "status": "ok", + "poller_alive": poller.alive, + "last_fetch": poller.last_fetch.isoformat() if poller.last_fetch else None, + "current_week_track_count": len(tracks), + } + + @app.get("/playlist") + def playlist(): + show = _current_show() + tracks = db.get_show_tracks(show.id) + return { + "show_id": show.id, + "week_start": show.week_start.isoformat(), + "week_end": show.week_end.isoformat(), + "tracks": tracks, + } + + @app.get("/playlist/{position}") + def playlist_track(position: int): + show = _current_show() + track = db.get_show_track_by_position(show.id, position) + if track is None: + raise HTTPException(status_code=404, detail=f"No track at position {position}") + return track + + @app.get("/shows") + def list_shows(limit: int = 20, offset: int = 0): + shows = db.list_shows(limit=limit, offset=offset) + return [ + { + "id": s.id, + "week_start": s.week_start.isoformat(), + "week_end": s.week_end.isoformat(), + "created_at": s.created_at.isoformat(), + } + for s in shows + ] + + @app.get("/shows/{show_id}") + def show_detail(show_id: int): + shows = db.list_shows(limit=1000, offset=0) + show = next((s for s in shows if s.id == show_id), None) + if show is None: + raise HTTPException(status_code=404, detail="Show not found") + tracks = db.get_show_tracks(show.id) + return { + "show_id": show.id, + "week_start": show.week_start.isoformat(), + "week_end": show.week_end.isoformat(), + "tracks": tracks, + } + + @app.post("/admin/refresh") + async def admin_refresh(body: RefreshRequest = RefreshRequest(), _=Depends(_require_admin)): + await poller.poll_once(full=body.full) + show = _current_show() + tracks = db.get_show_tracks(show.id) + return {"status": "refreshed", "track_count": len(tracks)} + + @app.post("/admin/tracks") + def admin_add_track(body: AddTrackRequest, _=Depends(_require_admin)): + show = _current_show() + if body.track_id is not None: + db.add_track_to_show(show.id, body.track_id, body.position) + return {"status": "added"} + raise HTTPException(status_code=400, detail="Provide track_id") + + @app.delete("/admin/tracks/{track_id}") + def admin_remove_track(track_id: int, _=Depends(_require_admin)): + show = _current_show() + if not db.has_track_in_show(show.id, track_id): + raise HTTPException(status_code=404, detail="Track not in current show") + db.remove_show_track(show.id, track_id) + return {"status": "removed"} + + @app.put("/admin/tracks/{track_id}/position") + def admin_move_track(track_id: int, body: MoveTrackRequest, _=Depends(_require_admin)): + show = _current_show() + if not db.has_track_in_show(show.id, track_id): + raise HTTPException(status_code=404, detail="Track not in current show") + db.move_show_track(show.id, track_id, body.position) + return {"status": "moved"} + + return app diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..f857aee --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,119 @@ +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi.testclient import TestClient + +from ntr_fetcher.api import create_app +from ntr_fetcher.db import Database +from ntr_fetcher.models import Track + +# Fixed "now" so seeded show (week_start Mar 12 2am–Mar 19 2am UTC) matches get_show_week +FAKE_NOW = datetime(2026, 3, 14, 12, 0, 0, tzinfo=timezone.utc) + + +@pytest.fixture +def db(tmp_path): + database = Database(str(tmp_path / "test.db")) + database.initialize() + return database + + +@pytest.fixture +def app(db): + poller = MagicMock() + poller.last_fetch = datetime(2026, 3, 12, 12, 0, 0, tzinfo=timezone.utc) + poller.alive = True + poller.poll_once = AsyncMock() + return create_app(db=db, poller=poller, admin_token="test-token") + + +@pytest.fixture +def client(app): + with patch("ntr_fetcher.api.datetime") as mock_dt: + mock_dt.now.return_value = FAKE_NOW + yield TestClient(app) + + +def _seed_show(db): + week_start = datetime(2026, 3, 12, 2, 0, 0, tzinfo=timezone.utc) + week_end = datetime(2026, 3, 19, 2, 0, 0, tzinfo=timezone.utc) + show = db.get_or_create_show(week_start, week_end) + t1 = Track(1, "Song A", "Artist A", "https://soundcloud.com/a/1", None, 180000, "cc-by", + datetime(2026, 3, 14, 1, 0, 0, tzinfo=timezone.utc), "{}") + t2 = Track(2, "Song B", "Artist B", "https://soundcloud.com/b/2", None, 200000, "cc-by-sa", + datetime(2026, 3, 14, 2, 0, 0, tzinfo=timezone.utc), "{}") + db.upsert_track(t1) + db.upsert_track(t2) + db.set_show_tracks(show.id, [t1.id, t2.id]) + return show + + +def test_health(client): + resp = client.get("/health") + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "ok" + assert data["poller_alive"] is True + + +def test_playlist(client, db): + _seed_show(db) + resp = client.get("/playlist") + assert resp.status_code == 200 + data = resp.json() + assert len(data["tracks"]) == 2 + assert data["tracks"][0]["position"] == 1 + assert data["tracks"][0]["title"] == "Song A" + + +def test_playlist_by_position(client, db): + _seed_show(db) + resp = client.get("/playlist/2") + assert resp.status_code == 200 + assert resp.json()["title"] == "Song B" + + +def test_playlist_by_position_not_found(client, db): + _seed_show(db) + resp = client.get("/playlist/99") + assert resp.status_code == 404 + + +def test_shows_list(client, db): + _seed_show(db) + resp = client.get("/shows") + assert resp.status_code == 200 + assert len(resp.json()) >= 1 + + +def test_shows_detail(client, db): + show = _seed_show(db) + resp = client.get(f"/shows/{show.id}") + assert resp.status_code == 200 + assert len(resp.json()["tracks"]) == 2 + + +def test_admin_refresh_requires_token(client): + resp = client.post("/admin/refresh") + assert resp.status_code == 401 + + +def test_admin_refresh_with_token(client): + resp = client.post( + "/admin/refresh", + headers={"Authorization": "Bearer test-token"}, + json={"full": False}, + ) + assert resp.status_code == 200 + + +def test_admin_remove_track(client, db): + show = _seed_show(db) + resp = client.delete( + "/admin/tracks/1", + headers={"Authorization": "Bearer test-token"}, + ) + assert resp.status_code == 200 + tracks = db.get_show_tracks(show.id) + assert len(tracks) == 1