bvandeusen/FabledCurator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 33 Wiki Activity

fc3e: PostFeedService — cursor-paginated unified posts feed mirroring GalleryService.scroll

Browse Source 
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-05-21 18:54:49 -04:00
parent 07edb73373
commit 5925ebe1f1
2 changed files with 543 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/post_feed_service.py
+209
View File
@@ -0,0 +1,209 @@
"""FC-3e: cursor-paginated read service for the Posts stream.
Mirrors GalleryService.scroll's cursor encoding so the frontend pattern
is identical: base64 of "<iso8601_sort_key>|<post_id>". Sort key is
COALESCE(Post.post_date, Post.downloaded_at) so posts without a
publish date sort by when we captured them.
Pure read-surface; no writes. The service composes the post dict
(including thumbnails from ImageRecord.primary_post_id and non-media
attachments from PostAttachment) so the API layer can jsonify directly.
"""
from __future__ import annotations
import base64
from datetime import datetime
from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from ..models import Artist, ImageRecord, Post, PostAttachment, Source
from ..utils.text import html_to_plain, truncate_at_word
from .gallery_service import thumbnail_url
CURSOR_SEPARATOR = "|"
DESCRIPTION_LIMIT = 280
THUMBNAIL_LIMIT = 6
def encode_cursor(sort_key: datetime, post_id: int) -> str:
raw = f"{sort_key.isoformat()}{CURSOR_SEPARATOR}{post_id}"
return base64.urlsafe_b64encode(raw.encode()).decode()
def decode_cursor(cursor: str) -> tuple[datetime, int]:
try:
raw = base64.urlsafe_b64decode(cursor.encode()).decode()
ts_part, id_part = raw.split(CURSOR_SEPARATOR, 1)
return datetime.fromisoformat(ts_part), int(id_part)
except Exception as exc:
raise ValueError(f"invalid cursor: {cursor!r}") from exc
def _sort_key():
"""Postgres COALESCE expression used in ORDER BY and WHERE clauses."""
return func.coalesce(Post.post_date, Post.downloaded_at)
class PostFeedService:
def __init__(self, session: AsyncSession):
self.session = session
async def scroll(
self,
*,
cursor: str | None = None,
artist_id: int | None = None,
platform: str | None = None,
limit: int = 24,
) -> dict:
if limit < 1 or limit > 100:
raise ValueError("limit must be between 1 and 100")
sort_key = _sort_key()
stmt = (
select(Post, Artist, Source)
.join(Source, Post.source_id == Source.id)
.join(Artist, Source.artist_id == Artist.id)
)
if artist_id is not None:
stmt = stmt.where(Source.artist_id == artist_id)
if platform is not None:
stmt = stmt.where(Source.platform == platform)
if cursor:
cur_ts, cur_id = decode_cursor(cursor)
stmt = stmt.where(
or_(
sort_key < cur_ts,
and_(sort_key == cur_ts, Post.id < cur_id),
)
)
stmt = stmt.order_by(sort_key.desc(), Post.id.desc()).limit(limit + 1)
rows = (await self.session.execute(stmt)).all()
next_cursor: str | None = None
if len(rows) > limit:
last_post, _, _ = rows[limit - 1]
last_key = last_post.post_date or last_post.downloaded_at
next_cursor = encode_cursor(last_key, last_post.id)
rows = rows[:limit]
post_ids = [p.id for p, _, _ in rows]
thumbs_map = await self._thumbnails_for(post_ids)
atts_map = await self._attachments_for(post_ids)
items = [
self._to_dict(post, artist, source, thumbs_map, atts_map)
for post, artist, source in rows
]
return {"items": items, "next_cursor": next_cursor}
async def get_post(self, post_id: int) -> dict | None:
row = (await self.session.execute(
select(Post, Artist, Source)
.join(Source, Post.source_id == Source.id)
.join(Artist, Source.artist_id == Artist.id)
.where(Post.id == post_id)
)).one_or_none()
if row is None:
return None
post, artist, source = row
thumbs_map = await self._thumbnails_for([post.id])
atts_map = await self._attachments_for([post.id])
item = self._to_dict(post, artist, source, thumbs_map, atts_map)
item["description_full"] = html_to_plain(post.description)
return item
# --- composition helpers ---------------------------------------------
async def _thumbnails_for(self, post_ids: list[int]) -> dict[int, dict]:
"""post_id -> {"thumbs": [...up to 6], "more": int}.
Selects THUMBNAIL_LIMIT+1 images per post via window function so we
can detect overflow in a single query.
"""
if not post_ids:
return {}
# Rank images within each post and fetch only the top THUMBNAIL_LIMIT+1.
ranked = (
select(
ImageRecord.id,
ImageRecord.primary_post_id,
ImageRecord.sha256,
ImageRecord.mime,
func.row_number().over(
partition_by=ImageRecord.primary_post_id,
order_by=ImageRecord.id.asc(),
).label("rn"),
func.count(ImageRecord.id).over(
partition_by=ImageRecord.primary_post_id,
).label("total"),
)
.where(ImageRecord.primary_post_id.in_(post_ids))
.subquery()
)
rows = (await self.session.execute(
select(
ranked.c.id, ranked.c.primary_post_id,
ranked.c.sha256, ranked.c.mime, ranked.c.total,
).where(ranked.c.rn <= THUMBNAIL_LIMIT)
)).all()
out: dict[int, dict] = {pid: {"thumbs": [], "more": 0} for pid in post_ids}
for img_id, pid, sha, mime, total in rows:
entry = out.setdefault(pid, {"thumbs": [], "more": 0})
entry["thumbs"].append({
"image_id": img_id,
"thumbnail_url": thumbnail_url(sha, mime),
"mime": mime,
})
# `total` is constant per partition; overflow = total - THUMBNAIL_LIMIT.
entry["more"] = max(0, total - THUMBNAIL_LIMIT)
return out
async def _attachments_for(self, post_ids: list[int]) -> dict[int, list[dict]]:
if not post_ids:
return {}
rows = (await self.session.execute(
select(PostAttachment)
.where(PostAttachment.post_id.in_(post_ids))
.order_by(PostAttachment.id.asc())
)).scalars().all()
out: dict[int, list[dict]] = {pid: [] for pid in post_ids}
for att in rows:
out.setdefault(att.post_id, []).append({
"id": att.id,
"original_filename": att.original_filename,
"ext": att.ext,
"mime": att.mime,
"size_bytes": att.size_bytes,
"download_url": f"/api/attachments/{att.id}/download",
})
return out
def _to_dict(
self, post: Post, artist: Artist, source: Source,
thumbs_map: dict, atts_map: dict,
) -> dict:
plain_full = html_to_plain(post.description) if post.description else None
if plain_full is None:
description_plain, truncated = None, False
else:
description_plain, truncated = truncate_at_word(plain_full, DESCRIPTION_LIMIT)
thumbs_entry = thumbs_map.get(post.id, {"thumbs": [], "more": 0})
return {
"id": post.id,
"external_post_id": post.external_post_id,
"post_url": post.post_url,
"post_title": post.post_title,
"post_date": post.post_date.isoformat() if post.post_date else None,
"downloaded_at": post.downloaded_at.isoformat(),
"description_plain": description_plain,
"description_truncated": truncated,
"artist": {"id": artist.id, "name": artist.name, "slug": artist.slug},
"source": {"id": source.id, "platform": source.platform},
"thumbnails": thumbs_entry["thumbs"],
"thumbnails_more": thumbs_entry["more"],
"attachments": atts_map.get(post.id, []),
}
tests/test_post_feed_service.py
+334
View File
@@ -0,0 +1,334 @@
"""FC-3e: PostFeedService unit tests.
Covers cursor encode/decode, ordering (coalesce(post_date,
downloaded_at) DESC, id DESC), artist/platform filters, pagination
boundary, and the thumbnails/attachments composition.
"""
from datetime import UTC, datetime, timedelta
import pytest
from backend.app.models import (
Artist,
ImageRecord,
Post,
PostAttachment,
Source,
)
from backend.app.services.post_feed_service import (
PostFeedService,
decode_cursor,
encode_cursor,
)
pytestmark = pytest.mark.integration
# --- Cursor round-trip ----------------------------------------------------
def test_cursor_round_trip():
ts = datetime(2026, 5, 21, 12, 34, 56, tzinfo=UTC)
c = encode_cursor(ts, 42)
assert decode_cursor(c) == (ts, 42)
def test_decode_cursor_rejects_garbage():
with pytest.raises(ValueError):
decode_cursor("not-base64!!!")
# --- Fixture builders -----------------------------------------------------
async def _seed_artist(db, name: str):
a = Artist(name=name, slug=name.lower().replace(" ", "-"))
db.add(a)
await db.flush()
return a
async def _seed_source(db, artist_id: int, platform: str, url: str):
s = Source(artist_id=artist_id, platform=platform, url=url, enabled=True)
db.add(s)
await db.flush()
return s
async def _seed_post(
db, source_id: int, *, external_id: str,
post_date=None, downloaded_at=None, title=None, description=None,
):
p = Post(
source_id=source_id, external_post_id=external_id,
post_title=title, post_date=post_date,
description=description,
)
if downloaded_at is not None:
p.downloaded_at = downloaded_at
db.add(p)
await db.flush()
return p
# --- scroll() ordering and pagination ------------------------------------
@pytest.mark.asyncio
async def test_scroll_orders_by_coalesce_post_date_then_downloaded_at(db):
artist = await _seed_artist(db, "alice-order")
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-order")
now = datetime.now(UTC)
# post A: post_date older than its downloaded_at; sort key = post_date
a = await _seed_post(
db, src.id, external_id="A",
post_date=now - timedelta(days=10),
downloaded_at=now,
)
# post B: post_date NULL → sort key = downloaded_at
b = await _seed_post(
db, src.id, external_id="B",
post_date=None,
downloaded_at=now - timedelta(days=5),
)
# post C: post_date newest of all
c = await _seed_post(
db, src.id, external_id="C",
post_date=now - timedelta(days=1),
downloaded_at=now - timedelta(days=20),
)
await db.commit()
page = await PostFeedService(db).scroll(cursor=None, limit=10)
ids = [item["id"] for item in page["items"]]
# Expected order by coalesce(post_date, downloaded_at) DESC:
# C (1 day ago) > B (5 days ago via downloaded_at) > A (10 days ago)
assert ids == [c.id, b.id, a.id]
assert page["next_cursor"] is None # 3 items, limit 10, no more
@pytest.mark.asyncio
async def test_scroll_pagination_across_two_pages(db):
artist = await _seed_artist(db, "alice-page")
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-page")
now = datetime.now(UTC)
posts = []
for i in range(5):
p = await _seed_post(
db, src.id, external_id=f"P{i}",
post_date=now - timedelta(days=i),
)
posts.append(p)
await db.commit()
page1 = await PostFeedService(db).scroll(cursor=None, limit=2)
assert len(page1["items"]) == 2
assert page1["next_cursor"] is not None
assert [it["id"] for it in page1["items"]] == [posts[0].id, posts[1].id]
page2 = await PostFeedService(db).scroll(cursor=page1["next_cursor"], limit=2)
assert [it["id"] for it in page2["items"]] == [posts[2].id, posts[3].id]
assert page2["next_cursor"] is not None
page3 = await PostFeedService(db).scroll(cursor=page2["next_cursor"], limit=2)
assert [it["id"] for it in page3["items"]] == [posts[4].id]
assert page3["next_cursor"] is None
# --- Filters --------------------------------------------------------------
@pytest.mark.asyncio
async def test_scroll_filters_by_artist(db):
alice = await _seed_artist(db, "alice-filter")
bob = await _seed_artist(db, "bob-filter")
src_a = await _seed_source(db, alice.id, "patreon", "https://p/alice-f")
src_b = await _seed_source(db, bob.id, "patreon", "https://p/bob-f")
now = datetime.now(UTC)
pa = await _seed_post(db, src_a.id, external_id="PA", post_date=now)
await _seed_post(db, src_b.id, external_id="PB", post_date=now)
await db.commit()
page = await PostFeedService(db).scroll(
cursor=None, limit=10, artist_id=alice.id,
)
assert [it["id"] for it in page["items"]] == [pa.id]
@pytest.mark.asyncio
async def test_scroll_filters_by_platform(db):
artist = await _seed_artist(db, "alice-platf")
src_p = await _seed_source(db, artist.id, "patreon", "https://p/alice-pp")
src_d = await _seed_source(db, artist.id, "deviantart", "https://d/alice-dd")
now = datetime.now(UTC)
pp = await _seed_post(db, src_p.id, external_id="PP", post_date=now)
await _seed_post(db, src_d.id, external_id="PD", post_date=now)
await db.commit()
page = await PostFeedService(db).scroll(
cursor=None, limit=10, platform="patreon",
)
assert [it["id"] for it in page["items"]] == [pp.id]
@pytest.mark.asyncio
async def test_scroll_combined_artist_and_platform(db):
alice = await _seed_artist(db, "alice-combo")
bob = await _seed_artist(db, "bob-combo")
src_alice_patreon = await _seed_source(db, alice.id, "patreon", "https://p/alice-c")
src_alice_da = await _seed_source(db, alice.id, "deviantart", "https://d/alice-c")
src_bob_patreon = await _seed_source(db, bob.id, "patreon", "https://p/bob-c")
now = datetime.now(UTC)
target = await _seed_post(
db, src_alice_patreon.id, external_id="TARGET", post_date=now,
)
# Distractors — wrong platform for alice, wrong artist for patreon.
await _seed_post(db, src_alice_da.id, external_id="DA", post_date=now)
await _seed_post(db, src_bob_patreon.id, external_id="BP", post_date=now)
await db.commit()
page = await PostFeedService(db).scroll(
cursor=None, limit=10, artist_id=alice.id, platform="patreon",
)
assert [it["id"] for it in page["items"]] == [target.id]
# --- Dict shape and content ----------------------------------------------
@pytest.mark.asyncio
async def test_scroll_item_shape_minimal(db):
artist = await _seed_artist(db, "alice-shape")
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-sh")
now = datetime.now(UTC)
await _seed_post(
db, src.id, external_id="SHAPE", post_date=now,
title="Hello", description="<p>world</p>",
)
await db.commit()
page = await PostFeedService(db).scroll(cursor=None, limit=10)
item = page["items"][0]
assert set(item.keys()) >= {
"id", "external_post_id", "post_url", "post_title",
"post_date", "downloaded_at",
"description_plain", "description_truncated",
"artist", "source",
"thumbnails", "thumbnails_more", "attachments",
}
assert item["post_title"] == "Hello"
assert item["description_plain"] == "world"
assert item["description_truncated"] is False
assert item["artist"]["slug"] == "alice-shape"
assert item["source"]["platform"] == "patreon"
# List response NEVER carries the full description.
assert "description_full" not in item
@pytest.mark.asyncio
async def test_scroll_description_truncates_long_text(db):
artist = await _seed_artist(db, "alice-long")
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-lo")
long_text = "word " * 200 # ~1000 chars
await _seed_post(
db, src.id, external_id="LONG",
post_date=datetime.now(UTC),
description=long_text,
)
await db.commit()
page = await PostFeedService(db).scroll(cursor=None, limit=10)
item = page["items"][0]
assert item["description_truncated"] is True
assert len(item["description_plain"]) <= 281 # 280 + ellipsis char
# --- Thumbnails composition (primary_post_id linkage) --------------------
@pytest.mark.asyncio
async def test_scroll_thumbnails_show_first_six_with_overflow_count(db):
artist = await _seed_artist(db, "alice-thumbs")
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-th")
post = await _seed_post(
db, src.id, external_id="THUMB",
post_date=datetime.now(UTC),
)
# Seed 8 ImageRecords linked to this post (unique sha + path each).
for i in range(8):
db.add(ImageRecord(
path=f"/images/thumb-{post.id}-{i}.jpg",
sha256=f"{post.id:02d}" + f"{i:062d}",
size_bytes=10, mime="image/jpeg", width=10, height=10,
origin="downloaded", primary_post_id=post.id,
artist_id=artist.id,
))
await db.commit()
page = await PostFeedService(db).scroll(cursor=None, limit=10)
item = page["items"][0]
assert len(item["thumbnails"]) == 6
assert item["thumbnails_more"] == 2
for t in item["thumbnails"]:
assert "image_id" in t
assert t["thumbnail_url"].startswith("/images/thumbs/")
assert t["mime"] == "image/jpeg"
# --- Attachments composition ---------------------------------------------
@pytest.mark.asyncio
async def test_scroll_attachments_returns_non_media_files(db):
artist = await _seed_artist(db, "alice-att")
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-at")
post = await _seed_post(
db, src.id, external_id="ATT", post_date=datetime.now(UTC),
)
db.add(PostAttachment(
post_id=post.id, artist_id=artist.id,
sha256="a" * 64, path="/images/att/x.pdf",
original_filename="notes.pdf", ext="pdf",
mime="application/pdf", size_bytes=12345,
))
await db.commit()
page = await PostFeedService(db).scroll(cursor=None, limit=10)
item = page["items"][0]
assert len(item["attachments"]) == 1
att = item["attachments"][0]
assert att["original_filename"] == "notes.pdf"
assert att["ext"] == "pdf"
assert att["mime"] == "application/pdf"
assert att["size_bytes"] == 12345
assert att["download_url"] == f"/api/attachments/{att['id']}/download"
# --- get_post() detail ---------------------------------------------------
@pytest.mark.asyncio
async def test_get_post_returns_full_description(db):
artist = await _seed_artist(db, "alice-detail")
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-de")
long_text = "word " * 200
p = await _seed_post(
db, src.id, external_id="DETAIL",
post_date=datetime.now(UTC),
description=long_text,
)
await db.commit()
item = await PostFeedService(db).get_post(p.id)
assert item is not None
assert "description_full" in item
# Full text is HTML-stripped but NOT truncated.
assert len(item["description_full"]) > 280
@pytest.mark.asyncio
async def test_get_post_unknown_returns_none(db):
item = await PostFeedService(db).get_post(99999)
assert item is None