feat: M2 (part 1) — subscription import, YouTube client, quota guard

- Channel/Subscription/Video/ApiQuotaUsage models + migration 0002
- Synchronous YouTube Data API client with OAuth token refresh and per-call
  quota accounting (API key preferred for public reads when configured)
- Subscription import: upsert channels + subscription links, prune unsubscribed,
  fetch channel details (uploads playlist, topics, stats, handle, country)
- Central quota guard tracking units per US-Pacific day against a daily budget
- POST /api/sync/subscriptions and GET /api/sync/status endpoints
This commit is contained in:
npeter83 2026-06-11 01:22:07 +02:00
parent 3a774cf7d6
commit 24b6e0026d
10 changed files with 603 additions and 3 deletions

View file

@ -0,0 +1,127 @@
"""ingest core: channels, subscriptions, videos, api_quota_usage
Revision ID: 0002_ingest_core
Revises: 0001_initial
Create Date: 2026-06-11
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "0002_ingest_core"
down_revision: Union[str, None] = "0001_initial"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"channels",
sa.Column("id", sa.String(length=32), primary_key=True),
sa.Column("title", sa.String(length=255), nullable=True),
sa.Column("handle", sa.String(length=255), nullable=True),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("thumbnail_url", sa.String(length=1024), nullable=True),
sa.Column("uploads_playlist_id", sa.String(length=64), nullable=True),
sa.Column("subscriber_count", sa.BigInteger(), nullable=True),
sa.Column("video_count", sa.BigInteger(), nullable=True),
sa.Column("topic_categories", sa.JSON(), nullable=True),
sa.Column("default_language", sa.String(length=16), nullable=True),
sa.Column("country", sa.String(length=8), nullable=True),
sa.Column("details_synced_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("recent_synced_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_rss_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("backfill_cursor", sa.String(length=128), nullable=True),
sa.Column("backfill_done", sa.Boolean(), nullable=False, server_default="false"),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
)
op.create_table(
"subscriptions",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column(
"user_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"channel_id",
sa.String(length=32),
sa.ForeignKey("channels.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("yt_subscription_id", sa.String(length=128), nullable=True),
sa.Column("priority", sa.Integer(), nullable=False, server_default="0"),
sa.Column("hidden", sa.Boolean(), nullable=False, server_default="false"),
sa.Column("subscribed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),
)
op.create_index("ix_subscriptions_user_id", "subscriptions", ["user_id"])
op.create_index("ix_subscriptions_channel_id", "subscriptions", ["channel_id"])
op.create_table(
"videos",
sa.Column("id", sa.String(length=16), primary_key=True),
sa.Column(
"channel_id",
sa.String(length=32),
sa.ForeignKey("channels.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("title", sa.Text(), nullable=True),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("published_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("thumbnail_url", sa.String(length=1024), nullable=True),
sa.Column("duration_seconds", sa.Integer(), nullable=True),
sa.Column("view_count", sa.BigInteger(), nullable=True),
sa.Column("like_count", sa.BigInteger(), nullable=True),
sa.Column("category_id", sa.Integer(), nullable=True),
sa.Column("topic_categories", sa.JSON(), nullable=True),
sa.Column("default_language", sa.String(length=16), nullable=True),
sa.Column("detected_language", sa.String(length=16), nullable=True),
sa.Column("is_short", sa.Boolean(), nullable=False, server_default="false"),
sa.Column("live_status", sa.String(length=16), nullable=False, server_default="none"),
sa.Column("enriched_at", sa.DateTime(timezone=True), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
)
op.create_index("ix_videos_channel_id", "videos", ["channel_id"])
op.create_index("ix_videos_published_at", "videos", ["published_at"])
op.create_index("ix_videos_is_short", "videos", ["is_short"])
op.create_index("ix_videos_live_status", "videos", ["live_status"])
op.create_table(
"api_quota_usage",
sa.Column("day", sa.Date(), primary_key=True),
sa.Column("units_used", sa.Integer(), nullable=False, server_default="0"),
)
def downgrade() -> None:
op.drop_table("api_quota_usage")
op.drop_index("ix_videos_live_status", table_name="videos")
op.drop_index("ix_videos_is_short", table_name="videos")
op.drop_index("ix_videos_published_at", table_name="videos")
op.drop_index("ix_videos_channel_id", table_name="videos")
op.drop_table("videos")
op.drop_index("ix_subscriptions_channel_id", table_name="subscriptions")
op.drop_index("ix_subscriptions_user_id", table_name="subscriptions")
op.drop_table("subscriptions")
op.drop_table("channels")

View file

@ -24,6 +24,16 @@ class Settings(BaseSettings):
# Origin of a separately served frontend dev server (enables CORS). Empty in production.
frontend_origin: str = ""
# --- Sync / YouTube Data API ---
# Optional API key for public reads (channels/videos/playlistItems). When set it is
# preferred for shared backfill/enrichment so it doesn't depend on a specific user.
youtube_api_key: str = ""
# Daily quota budget (units). Default leaves headroom under the 10,000/day free limit.
quota_daily_budget: int = 9000
# Recent-first backfill: how far back to fetch on the first pass per channel.
backfill_recent_max_videos: int = 200
backfill_recent_max_days: int = 365
@property
def allowed_email_set(self) -> set[str]:
return {e.strip().lower() for e in self.allowed_emails.split(",") if e.strip()}

View file

@ -8,7 +8,7 @@ from starlette.middleware.sessions import SessionMiddleware
from app import auth
from app.config import settings
from app.routes import health
from app.routes import health, sync
app = FastAPI(title=settings.app_name)
@ -30,6 +30,7 @@ if settings.frontend_origin:
app.include_router(health.router)
app.include_router(auth.router)
app.include_router(sync.router)
STATIC_DIR = Path(__file__).parent / "static"
app.mount("/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="assets")

View file

@ -1,6 +1,18 @@
from datetime import datetime
from datetime import date, datetime
from sqlalchemy import DateTime, ForeignKey, String, func
from sqlalchemy import (
BigInteger,
Boolean,
Date,
DateTime,
ForeignKey,
Integer,
JSON,
String,
Text,
UniqueConstraint,
func,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db import Base
@ -22,6 +34,9 @@ class User(Base):
token: Mapped["OAuthToken | None"] = relationship(
back_populates="user", uselist=False, cascade="all, delete-orphan"
)
subscriptions: Mapped[list["Subscription"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
class OAuthToken(Base):
@ -41,3 +56,110 @@ class OAuthToken(Base):
)
user: Mapped["User"] = relationship(back_populates="token")
class Channel(Base):
"""A YouTube channel. Shared across all users (one channel's videos are the same
for everyone), so its expensive metadata is fetched and stored only once."""
__tablename__ = "channels"
id: Mapped[str] = mapped_column(String(32), primary_key=True) # "UC..." channel id
title: Mapped[str | None] = mapped_column(String(255))
handle: Mapped[str | None] = mapped_column(String(255))
description: Mapped[str | None] = mapped_column(Text)
thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
uploads_playlist_id: Mapped[str | None] = mapped_column(String(64))
subscriber_count: Mapped[int | None] = mapped_column(BigInteger)
video_count: Mapped[int | None] = mapped_column(BigInteger)
topic_categories: Mapped[list | None] = mapped_column(JSON)
default_language: Mapped[str | None] = mapped_column(String(16))
country: Mapped[str | None] = mapped_column(String(8))
# Sync bookkeeping.
details_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
recent_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
last_rss_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
backfill_cursor: Mapped[str | None] = mapped_column(String(128))
backfill_done: Mapped[bool] = mapped_column(
Boolean, default=False, server_default="false"
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
videos: Mapped[list["Video"]] = relationship(back_populates="channel")
subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel")
class Subscription(Base):
"""Per-user link to a channel, with the user's own priority/visibility overrides."""
__tablename__ = "subscriptions"
__table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), index=True
)
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
# YouTube's subscription resource id, needed to unsubscribe via the API.
yt_subscription_id: Mapped[str | None] = mapped_column(String(128))
priority: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
hidden: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
subscribed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
user: Mapped["User"] = relationship(back_populates="subscriptions")
channel: Mapped["Channel"] = relationship(back_populates="subscriptions")
class Video(Base):
"""A YouTube video. Shared across users; per-user state lives elsewhere."""
__tablename__ = "videos"
id: Mapped[str] = mapped_column(String(16), primary_key=True) # video id
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
title: Mapped[str | None] = mapped_column(Text)
description: Mapped[str | None] = mapped_column(Text)
published_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), index=True
)
thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
duration_seconds: Mapped[int | None] = mapped_column(Integer)
view_count: Mapped[int | None] = mapped_column(BigInteger)
like_count: Mapped[int | None] = mapped_column(BigInteger)
category_id: Mapped[int | None] = mapped_column(Integer)
topic_categories: Mapped[list | None] = mapped_column(JSON)
default_language: Mapped[str | None] = mapped_column(String(16))
detected_language: Mapped[str | None] = mapped_column(String(16))
is_short: Mapped[bool] = mapped_column(
Boolean, default=False, server_default="false", index=True
)
# none | live | upcoming | premiere | was_live
live_status: Mapped[str] = mapped_column(
String(16), default="none", server_default="none", index=True
)
enriched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
channel: Mapped["Channel"] = relationship(back_populates="videos")
class ApiQuotaUsage(Base):
"""Tracks YouTube Data API units spent per Pacific-time day (the quota resets at
midnight Pacific). The whole app shares one daily budget."""
__tablename__ = "api_quota_usage"
day: Mapped[date] = mapped_column(Date, primary_key=True)
units_used: Mapped[int] = mapped_column(Integer, default=0, server_default="0")

50
backend/app/quota.py Normal file
View file

@ -0,0 +1,50 @@
"""Central YouTube Data API quota guard.
The whole app shares one daily budget (the API resets at midnight US Pacific time).
Steady-state work (RSS detection is free; enrichment of new videos is cheap) should
always be allowed; expensive backfill must yield to the remaining budget.
"""
from datetime import datetime
from zoneinfo import ZoneInfo
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session
from app.config import settings
from app.models import ApiQuotaUsage
_PACIFIC = ZoneInfo("America/Los_Angeles")
def pacific_today():
return datetime.now(_PACIFIC).date()
def units_used_today(db: Session) -> int:
row = db.get(ApiQuotaUsage, pacific_today())
return row.units_used if row else 0
def remaining_today(db: Session) -> int:
return max(0, settings.quota_daily_budget - units_used_today(db))
def can_spend(db: Session, units: int) -> bool:
return remaining_today(db) >= units
def record_usage(db: Session, units: int) -> None:
"""Atomically add `units` to today's counter (upsert)."""
if units <= 0:
return
day = pacific_today()
stmt = (
pg_insert(ApiQuotaUsage)
.values(day=day, units_used=units)
.on_conflict_do_update(
index_elements=[ApiQuotaUsage.day],
set_={"units_used": ApiQuotaUsage.units_used + units},
)
)
db.execute(stmt)
db.commit()

View file

@ -0,0 +1,40 @@
from fastapi import APIRouter, Depends
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app import quota
from app.auth import current_user
from app.db import get_db
from app.models import Channel, Subscription, User, Video
from app.sync.subscriptions import import_subscriptions
router = APIRouter(prefix="/api/sync", tags=["sync"])
@router.post("/subscriptions")
def sync_subscriptions(
user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
before = quota.units_used_today(db)
result = import_subscriptions(db, user)
result["quota_used_estimate"] = quota.units_used_today(db) - before
result["quota_remaining_today"] = quota.remaining_today(db)
return result
@router.get("/status")
def sync_status(
user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
sub_count = db.scalar(
select(func.count())
.select_from(Subscription)
.where(Subscription.user_id == user.id)
)
return {
"subscriptions": sub_count,
"channels_total": db.scalar(select(func.count()).select_from(Channel)),
"videos_total": db.scalar(select(func.count()).select_from(Video)),
"quota_used_today": quota.units_used_today(db),
"quota_remaining_today": quota.remaining_today(db),
}

View file

View file

@ -0,0 +1,115 @@
"""Import the authenticated user's YouTube subscriptions and channel metadata."""
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import Channel, Subscription, User
from app.youtube.client import YouTubeClient, best_thumbnail
def _to_int(value) -> int | None:
try:
return int(value) if value is not None else None
except (TypeError, ValueError):
return None
def apply_channel_details(db: Session, items: list[dict]) -> None:
now = datetime.now(timezone.utc)
for item in items:
channel = db.get(Channel, item["id"])
if channel is None:
continue
snippet = item.get("snippet", {})
content = item.get("contentDetails", {})
stats = item.get("statistics", {})
topics = item.get("topicDetails", {})
channel.title = snippet.get("title") or channel.title
channel.handle = snippet.get("customUrl")
channel.description = snippet.get("description")
channel.thumbnail_url = (
best_thumbnail(snippet.get("thumbnails")) or channel.thumbnail_url
)
channel.default_language = snippet.get("defaultLanguage")
channel.country = snippet.get("country")
channel.uploads_playlist_id = content.get("relatedPlaylists", {}).get("uploads")
channel.subscriber_count = _to_int(stats.get("subscriberCount"))
channel.video_count = _to_int(stats.get("videoCount"))
channel.topic_categories = topics.get("topicCategories")
channel.details_synced_at = now
def import_subscriptions(db: Session, user: User) -> dict:
"""Pull all subscriptions for `user`, upsert channels + subscription links, prune
channels the user has unsubscribed from on YouTube, and fetch channel details for
any channel we don't have details for yet."""
fetched: dict[str, dict] = {}
new_channels = 0
with YouTubeClient(db, user) as yt:
for sub in yt.iter_subscriptions():
cid = sub.get("channel_id")
if cid:
fetched[cid] = sub
for cid, sub in fetched.items():
channel = db.get(Channel, cid)
if channel is None:
channel = Channel(
id=cid,
title=sub.get("title"),
description=sub.get("description"),
thumbnail_url=sub.get("thumbnail_url"),
)
db.add(channel)
new_channels += 1
sub_row = db.execute(
select(Subscription).where(
Subscription.user_id == user.id, Subscription.channel_id == cid
)
).scalar_one_or_none()
if sub_row is None:
sub_row = Subscription(user_id=user.id, channel_id=cid)
db.add(sub_row)
sub_row.yt_subscription_id = sub.get("yt_subscription_id")
db.flush()
# Prune subscriptions removed on YouTube.
removed = 0
existing = (
db.execute(select(Subscription).where(Subscription.user_id == user.id))
.scalars()
.all()
)
for sub_row in existing:
if sub_row.channel_id not in fetched:
db.delete(sub_row)
removed += 1
db.commit()
# Fetch details for channels that don't have them yet.
need_details = (
db.execute(
select(Channel).where(
Channel.id.in_(fetched.keys()),
Channel.details_synced_at.is_(None),
)
)
.scalars()
.all()
)
detailed = 0
if need_details:
items = yt.get_channels([c.id for c in need_details])
apply_channel_details(db, items)
detailed = len(items)
db.commit()
return {
"subscriptions": len(fetched),
"channels_new": new_channels,
"channels_detailed": detailed,
"removed_stale": removed,
}

View file

View file

@ -0,0 +1,135 @@
"""Thin synchronous YouTube Data API v3 client bound to a user's OAuth token.
Each list call costs 1 quota unit and is recorded via the central quota guard.
Public reads (channels/videos/playlistItems) use the configured API key when available
so they don't depend on a specific user's token; subscriptions.list?mine=true always
uses OAuth.
"""
from collections.abc import Iterator
from datetime import datetime, timedelta, timezone
import httpx
from app import quota
from app.config import settings
from app.models import User
from app.security import decrypt
GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
API_BASE = "https://www.googleapis.com/youtube/v3"
class YouTubeError(Exception):
pass
def _chunks(items: list, size: int) -> Iterator[list]:
for i in range(0, len(items), size):
yield items[i : i + size]
def best_thumbnail(thumbnails: dict | None) -> str | None:
if not thumbnails:
return None
for key in ("maxres", "standard", "high", "medium", "default"):
if key in thumbnails and thumbnails[key].get("url"):
return thumbnails[key]["url"]
return None
class YouTubeClient:
def __init__(self, db, user: User):
self.db = db
self.user = user
self._http = httpx.Client(timeout=30.0)
def __enter__(self) -> "YouTubeClient":
return self
def __exit__(self, *exc) -> None:
self._http.close()
# --- auth ---
def _access_token(self) -> str:
tok = self.user.token
if tok is None:
raise YouTubeError("User has no stored OAuth token")
now = datetime.now(timezone.utc)
if tok.access_token and tok.expiry and tok.expiry > now + timedelta(seconds=60):
return tok.access_token
refresh = decrypt(tok.refresh_token_enc)
if not refresh:
raise YouTubeError("No refresh token; user must re-authenticate")
resp = self._http.post(
GOOGLE_TOKEN_URL,
data={
"client_id": settings.google_client_id,
"client_secret": settings.google_client_secret,
"refresh_token": refresh,
"grant_type": "refresh_token",
},
)
if resp.status_code != 200:
raise YouTubeError(f"Token refresh failed: {resp.status_code} {resp.text[:200]}")
data = resp.json()
tok.access_token = data["access_token"]
tok.expiry = now + timedelta(seconds=int(data.get("expires_in", 3600)))
self.db.add(tok)
self.db.commit()
return tok.access_token
# --- core request ---
def _get(self, path: str, params: dict, cost: int = 1, allow_key: bool = True) -> dict:
p = dict(params)
headers = {}
if allow_key and settings.youtube_api_key:
p["key"] = settings.youtube_api_key
else:
headers["Authorization"] = f"Bearer {self._access_token()}"
resp = self._http.get(f"{API_BASE}/{path}", params=p, headers=headers)
quota.record_usage(self.db, cost)
if resp.status_code != 200:
raise YouTubeError(f"GET {path} -> {resp.status_code}: {resp.text[:300]}")
return resp.json()
# --- endpoints ---
def iter_subscriptions(self) -> Iterator[dict]:
"""Yield the authenticated user's subscriptions (requires OAuth)."""
page_token = None
while True:
params = {
"part": "snippet",
"mine": "true",
"maxResults": 50,
"order": "alphabetical",
}
if page_token:
params["pageToken"] = page_token
data = self._get("subscriptions", params, cost=1, allow_key=False)
for item in data.get("items", []):
snippet = item.get("snippet", {})
resource = snippet.get("resourceId", {})
yield {
"channel_id": resource.get("channelId"),
"title": snippet.get("title"),
"description": snippet.get("description"),
"thumbnail_url": best_thumbnail(snippet.get("thumbnails")),
"yt_subscription_id": item.get("id"),
}
page_token = data.get("nextPageToken")
if not page_token:
break
def get_channels(self, channel_ids: list[str]) -> list[dict]:
items: list[dict] = []
for batch in _chunks(channel_ids, 50):
data = self._get(
"channels",
{
"part": "snippet,contentDetails,statistics,topicDetails",
"id": ",".join(batch),
"maxResults": 50,
},
)
items.extend(data.get("items", []))
return items