audio streaming app
plyr.fm
refactor: extract fetch_list_item_uris and hydrate_tracks_from_uris (#1277)
Add fetch_list_item_uris() to _internal/atproto/records/fm_plyr/list.py
— fetches an ATProto list record and returns ordered item URIs. Replaces
5 copy-pasted fetch-then-extract blocks across playlists, albums, and
recommendations.
Add hydrate_tracks_from_uris() to api/lists/hydration.py — loads tracks
by AT-URI, batch-aggregates like/comment counts, resolves liked state,
returns ordered TrackResponses. Collapses the identical ~35-line hydration
block duplicated between get_playlist and get_playlist_by_uri.
playlists.py: 952 → 843 lines. Six unused imports removed as a side
effect (the hydration helper absorbed them).
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
authored by
nate nowack
Claude Opus 4.6 (1M context)
and committed by
GitHub
8b154031
ec0d740b
+130
-187
+2
backend/src/backend/_internal/atproto/records/__init__.py
+2
backend/src/backend/_internal/atproto/records/__init__.py
···
9
9
create_list_record,
10
10
create_track_record,
11
11
delete_record_by_uri,
12
+
fetch_list_item_uris,
12
13
get_record_public,
13
14
get_record_public_resilient,
14
15
update_comment_record,
···
43
44
"create_teal_play_record",
44
45
"create_track_record",
45
46
"delete_record_by_uri",
47
+
"fetch_list_item_uris",
46
48
"get_record_public",
47
49
"get_record_public_resilient",
48
50
"update_comment_record",
+2
backend/src/backend/_internal/atproto/records/fm_plyr/__init__.py
+2
backend/src/backend/_internal/atproto/records/fm_plyr/__init__.py
···
8
8
from backend._internal.atproto.records.fm_plyr.list import (
9
9
build_list_record,
10
10
create_list_record,
11
+
fetch_list_item_uris,
11
12
update_list_record,
12
13
upsert_album_list_record,
13
14
upsert_liked_list_record,
···
32
33
"create_list_record",
33
34
"create_track_record",
34
35
"delete_record_by_uri",
36
+
"fetch_list_item_uris",
35
37
"get_record_public",
36
38
"get_record_public_resilient",
37
39
"update_comment_record",
+18
-1
backend/src/backend/_internal/atproto/records/fm_plyr/list.py
+18
-1
backend/src/backend/_internal/atproto/records/fm_plyr/list.py
···
6
6
7
7
from backend._internal import Session as AuthSession
8
8
from backend._internal.atproto.client import make_pds_request
9
-
from backend._internal.atproto.records.fm_plyr.track import update_record
9
+
from backend._internal.atproto.records.fm_plyr.track import (
10
+
get_record_public_resilient,
11
+
update_record,
12
+
)
10
13
from backend.config import settings
11
14
12
15
logger = logging.getLogger(__name__)
···
210
213
)
211
214
logger.info(f"created liked list record for {auth_session.did}: {uri}")
212
215
return uri, cid
216
+
217
+
218
+
async def fetch_list_item_uris(
219
+
record_uri: str,
220
+
pds_url: str | None = None,
221
+
) -> list[str]:
222
+
"""fetch an ATProto list record and return its item URIs in order.
223
+
224
+
raises on PDS fetch failure — callers handle errors their own way
225
+
(playlists raise HTTPException, albums fall back to created_at order).
226
+
"""
227
+
record_data, _ = await get_record_public_resilient(record_uri, pds_url)
228
+
items = record_data.get("value", {}).get("items", [])
229
+
return [uri for item in items if (uri := item.get("subject", {}).get("uri"))]
+3
-8
backend/src/backend/api/albums/listing.py
+3
-8
backend/src/backend/api/albums/listing.py
···
11
11
12
12
from backend._internal import Session as AuthSession
13
13
from backend._internal import get_optional_session
14
-
from backend._internal.atproto.records import get_record_public_resilient
14
+
from backend._internal.atproto.records import fetch_list_item_uris
15
15
from backend.models import Album, Artist, Track, TrackLike, get_db
16
16
from backend.schemas import TrackResponse
17
17
from backend.utilities.aggregations import (
···
155
155
ordered_tracks: list[Track] = []
156
156
if album.atproto_record_uri:
157
157
try:
158
-
record_data, _ = await get_record_public_resilient(
159
-
record_uri=album.atproto_record_uri,
160
-
pds_url=artist.pds_url,
158
+
track_uris = await fetch_list_item_uris(
159
+
album.atproto_record_uri, artist.pds_url
161
160
)
162
-
163
-
items = record_data.get("value", {}).get("items", [])
164
-
track_uris = [item.get("subject", {}).get("uri") for item in items]
165
-
track_uris = [uri for uri in track_uris if uri]
166
161
167
162
# build uri -> track map
168
163
track_by_uri = {t.atproto_record_uri: t for t in all_tracks}
+5
-9
backend/src/backend/api/albums/mutations.py
+5
-9
backend/src/backend/api/albums/mutations.py
···
15
15
from backend._internal import require_artist_profile
16
16
from backend._internal.atproto.records import (
17
17
delete_record_by_uri,
18
-
get_record_public_resilient,
18
+
fetch_list_item_uris,
19
19
)
20
20
from backend._internal.atproto.records.fm_plyr.list import (
21
21
update_list_record,
···
274
274
select(Artist).where(Artist.did == album.artist_did)
275
275
)
276
276
artist_for_pds = artist_lookup.scalar_one()
277
-
record_data, _ = await get_record_public_resilient(
278
-
record_uri=album.atproto_record_uri,
279
-
pds_url=artist_for_pds.pds_url,
277
+
existing_uris = await fetch_list_item_uris(
278
+
album.atproto_record_uri, artist_for_pds.pds_url
280
279
)
281
-
items = record_data.get("value", {}).get("items", [])
282
-
for i, item in enumerate(items):
283
-
uri = item.get("subject", {}).get("uri")
284
-
if uri:
285
-
preserved_position_by_uri[uri] = i
280
+
for i, uri in enumerate(existing_uris):
281
+
preserved_position_by_uri[uri] = i
286
282
except Exception as e:
287
283
logger.debug(
288
284
f"finalize_album: failed to fetch existing list for preserved "
+59
backend/src/backend/api/lists/hydration.py
+59
backend/src/backend/api/lists/hydration.py
···
1
+
"""shared track hydration for list-backed endpoints (playlists, liked lists)."""
2
+
3
+
from sqlalchemy import select
4
+
from sqlalchemy.ext.asyncio import AsyncSession
5
+
from sqlalchemy.orm import selectinload
6
+
7
+
from backend.models import Track, TrackLike
8
+
from backend.schemas import TrackResponse
9
+
from backend.utilities.aggregations import get_comment_counts, get_like_counts
10
+
11
+
12
+
async def hydrate_tracks_from_uris(
13
+
db: AsyncSession,
14
+
track_uris: list[str],
15
+
session_did: str | None = None,
16
+
) -> list[TrackResponse]:
17
+
"""load tracks by AT-URI, aggregate counts, and return ordered TrackResponses.
18
+
19
+
preserves the order of track_uris (ATProto list record order).
20
+
skips URIs that don't resolve to a track in the database.
21
+
"""
22
+
if not track_uris:
23
+
return []
24
+
25
+
track_result = await db.execute(
26
+
select(Track)
27
+
.options(selectinload(Track.artist), selectinload(Track.album_rel))
28
+
.where(Track.atproto_record_uri.in_(track_uris))
29
+
)
30
+
all_tracks = track_result.scalars().all()
31
+
track_by_uri = {t.atproto_record_uri: t for t in all_tracks}
32
+
33
+
track_ids = [t.id for t in all_tracks]
34
+
like_counts = await get_like_counts(db, track_ids) if track_ids else {}
35
+
comment_counts = await get_comment_counts(db, track_ids) if track_ids else {}
36
+
37
+
liked_track_ids: set[int] = set()
38
+
if session_did and track_ids:
39
+
liked_result = await db.execute(
40
+
select(TrackLike.track_id).where(
41
+
TrackLike.user_did == session_did,
42
+
TrackLike.track_id.in_(track_ids),
43
+
)
44
+
)
45
+
liked_track_ids = set(liked_result.scalars().all())
46
+
47
+
tracks: list[TrackResponse] = []
48
+
for uri in track_uris:
49
+
if uri in track_by_uri:
50
+
track = track_by_uri[uri]
51
+
track_response = await TrackResponse.from_track(
52
+
track,
53
+
liked_track_ids=liked_track_ids,
54
+
like_counts=like_counts,
55
+
comment_counts=comment_counts,
56
+
)
57
+
tracks.append(track_response)
58
+
59
+
return tracks
+18
-110
backend/src/backend/api/lists/playlists.py
+18
-110
backend/src/backend/api/lists/playlists.py
···
8
8
from fastapi import Depends, File, Form, HTTPException, Query, UploadFile
9
9
from sqlalchemy import select
10
10
from sqlalchemy.ext.asyncio import AsyncSession
11
-
from sqlalchemy.orm import selectinload
12
11
13
12
from backend._internal import Session as AuthSession
14
13
from backend._internal import get_oauth_client, get_optional_session, require_auth
···
18
17
_reconstruct_oauth_session,
19
18
create_list_record,
20
19
delete_record_by_uri,
21
-
get_record_public_resilient,
20
+
fetch_list_item_uris,
22
21
update_list_record,
23
22
)
24
23
from backend._internal.image_uploads import COVER_EXTENSIONS, process_image_upload
···
29
28
CollectionEvent,
30
29
Playlist,
31
30
Track,
32
-
TrackLike,
33
31
get_db,
34
32
)
35
-
from backend.schemas import DeletedResponse, TrackResponse
33
+
from backend.schemas import DeletedResponse
36
34
from backend.storage import storage
37
-
from backend.utilities.aggregations import get_comment_counts, get_like_counts
38
35
from backend.utilities.redis import get_async_redis_client
39
36
37
+
from .hydration import hydrate_tracks_from_uris
40
38
from .router import router
41
39
from .schemas import (
42
40
AddTrackRequest,
···
198
196
199
197
playlist, artist = row
200
198
201
-
# fetch ATProto record (public - no auth needed)
199
+
# fetch track URIs from ATProto list record
202
200
try:
203
-
record_data, _ = await get_record_public_resilient(
204
-
record_uri=playlist.atproto_record_uri,
205
-
pds_url=artist.pds_url,
201
+
track_uris = await fetch_list_item_uris(
202
+
playlist.atproto_record_uri, artist.pds_url
206
203
)
207
204
except RecordNotFound:
208
205
raise HTTPException(
···
213
210
status_code=500, detail=f"failed to fetch playlist record: {e}"
214
211
) from e
215
212
216
-
items = record_data.get("value", {}).get("items", [])
217
-
218
-
# extract track URIs in order
219
-
track_uris = [item.get("subject", {}).get("uri") for item in items]
220
-
track_uris = [u for u in track_uris if u]
221
-
222
-
# hydrate track metadata from database
223
-
tracks: list[TrackResponse] = []
224
-
if track_uris:
225
-
track_result = await db.execute(
226
-
select(Track)
227
-
.options(selectinload(Track.artist), selectinload(Track.album_rel))
228
-
.where(Track.atproto_record_uri.in_(track_uris))
229
-
)
230
-
all_tracks = track_result.scalars().all()
231
-
track_by_uri = {t.atproto_record_uri: t for t in all_tracks}
232
-
233
-
track_ids = [t.id for t in all_tracks]
234
-
like_counts = await get_like_counts(db, track_ids) if track_ids else {}
235
-
comment_counts = await get_comment_counts(db, track_ids) if track_ids else {}
236
-
237
-
liked_track_ids: set[int] = set()
238
-
if session and track_ids:
239
-
liked_result = await db.execute(
240
-
select(TrackLike.track_id).where(
241
-
TrackLike.user_did == session.did,
242
-
TrackLike.track_id.in_(track_ids),
243
-
)
244
-
)
245
-
liked_track_ids = set(liked_result.scalars().all())
246
-
247
-
for u in track_uris:
248
-
if u in track_by_uri:
249
-
track = track_by_uri[u]
250
-
track_response = await TrackResponse.from_track(
251
-
track,
252
-
liked_track_ids=liked_track_ids,
253
-
like_counts=like_counts,
254
-
comment_counts=comment_counts,
255
-
)
256
-
tracks.append(track_response)
213
+
tracks = await hydrate_tracks_from_uris(
214
+
db, track_uris, session_did=session.did if session else None
215
+
)
257
216
258
217
return PlaylistWithTracksResponse(
259
218
id=playlist.id,
···
324
283
325
284
playlist, artist = row
326
285
327
-
# fetch ATProto record (public - no auth needed)
286
+
# fetch track URIs from ATProto list record
328
287
try:
329
-
record_data, _ = await get_record_public_resilient(
330
-
record_uri=playlist.atproto_record_uri,
331
-
pds_url=artist.pds_url,
288
+
track_uris = await fetch_list_item_uris(
289
+
playlist.atproto_record_uri, artist.pds_url
332
290
)
333
291
except RecordNotFound:
334
292
raise HTTPException(
···
339
297
status_code=500, detail=f"failed to fetch playlist record: {e}"
340
298
) from e
341
299
342
-
items = record_data.get("value", {}).get("items", [])
343
-
344
-
# extract track URIs in order
345
-
track_uris = [item.get("subject", {}).get("uri") for item in items]
346
-
track_uris = [uri for uri in track_uris if uri]
347
-
348
-
# hydrate track metadata from database
349
-
tracks: list[TrackResponse] = []
350
-
if track_uris:
351
-
track_result = await db.execute(
352
-
select(Track)
353
-
.options(selectinload(Track.artist), selectinload(Track.album_rel))
354
-
.where(Track.atproto_record_uri.in_(track_uris))
355
-
)
356
-
all_tracks = track_result.scalars().all()
357
-
track_by_uri = {t.atproto_record_uri: t for t in all_tracks}
358
-
359
-
# get track IDs for aggregation queries
360
-
track_ids = [t.id for t in all_tracks]
361
-
like_counts = await get_like_counts(db, track_ids) if track_ids else {}
362
-
comment_counts = await get_comment_counts(db, track_ids) if track_ids else {}
363
-
364
-
# get authenticated user's likes if session available
365
-
liked_track_ids: set[int] = set()
366
-
if session:
367
-
if track_ids:
368
-
liked_result = await db.execute(
369
-
select(TrackLike.track_id).where(
370
-
TrackLike.user_did == session.did,
371
-
TrackLike.track_id.in_(track_ids),
372
-
)
373
-
)
374
-
liked_track_ids = set(liked_result.scalars().all())
375
-
376
-
# maintain ATProto ordering, skip unavailable tracks
377
-
for uri in track_uris:
378
-
if uri in track_by_uri:
379
-
track = track_by_uri[uri]
380
-
track_response = await TrackResponse.from_track(
381
-
track,
382
-
liked_track_ids=liked_track_ids,
383
-
like_counts=like_counts,
384
-
comment_counts=comment_counts,
385
-
)
386
-
tracks.append(track_response)
387
-
# else: track exists in PDS list but not in our database - skip it
300
+
tracks = await hydrate_tracks_from_uris(
301
+
db, track_uris, session_did=session.did if session else None
302
+
)
388
303
389
304
return PlaylistWithTracksResponse(
390
305
id=playlist.id,
···
848
763
except Exception as e:
849
764
logger.debug("redis cache miss/error for recommendations: %s", e)
850
765
851
-
# get track IDs from the playlist's ATProto record
766
+
# get track URIs from the playlist's ATProto list record
852
767
try:
853
-
record_data, _ = await get_record_public_resilient(
854
-
record_uri=playlist.atproto_record_uri,
855
-
pds_url=artist.pds_url,
768
+
track_uris = await fetch_list_item_uris(
769
+
playlist.atproto_record_uri, artist.pds_url
856
770
)
857
771
except Exception as e:
858
772
logger.warning("failed to fetch playlist record for recommendations: %s", e)
859
773
return unavailable
860
-
861
-
items = record_data.get("value", {}).get("items", [])
862
-
track_uris = [
863
-
item.get("subject", {}).get("uri") for item in items if item.get("subject")
864
-
]
865
-
track_uris = [uri for uri in track_uris if uri]
866
774
867
775
if not track_uris:
868
776
return unavailable
+11
-22
backend/tests/api/test_albums.py
+11
-22
backend/tests/api/test_albums.py
···
522
522
db_session.add_all([track1, track2, track3])
523
523
await db_session.commit()
524
524
525
-
# mock ATProto record fetch to return custom order: track2, track3, track1
525
+
# mock ATProto list fetch to return custom order: track2, track3, track1
526
526
# (different from created_at order which would be track3, track2, track1)
527
-
mock_record = {
528
-
"value": {
529
-
"items": [
530
-
{"subject": {"uri": track2.atproto_record_uri, "cid": "cid2"}},
531
-
{"subject": {"uri": track3.atproto_record_uri, "cid": "cid3"}},
532
-
{"subject": {"uri": track1.atproto_record_uri, "cid": "cid1"}},
533
-
]
534
-
}
535
-
}
527
+
mock_uris = [
528
+
track2.atproto_record_uri,
529
+
track3.atproto_record_uri,
530
+
track1.atproto_record_uri,
531
+
]
536
532
537
533
with patch(
538
-
"backend.api.albums.listing.get_record_public_resilient",
534
+
"backend.api.albums.listing.fetch_list_item_uris",
539
535
new_callable=AsyncMock,
540
-
return_value=(mock_record, None),
536
+
return_value=mock_uris,
541
537
):
542
538
async with AsyncClient(
543
539
transport=ASGITransport(app=test_app), base_url="http://test"
···
1447
1443
new1_id = new1.id
1448
1444
1449
1445
# simulate the current list record having old1, old2 in that order
1450
-
existing_list_record = {
1451
-
"value": {
1452
-
"items": [
1453
-
{"subject": {"uri": old1.atproto_record_uri, "cid": "cidOld1"}},
1454
-
{"subject": {"uri": old2.atproto_record_uri, "cid": "cidOld2"}},
1455
-
]
1456
-
}
1457
-
}
1446
+
existing_uris = [old1.atproto_record_uri, old2.atproto_record_uri]
1458
1447
1459
1448
captured: dict[str, object] = {}
1460
1449
···
1472
1461
1473
1462
with (
1474
1463
patch(
1475
-
"backend.api.albums.mutations.get_record_public_resilient",
1464
+
"backend.api.albums.mutations.fetch_list_item_uris",
1476
1465
new_callable=AsyncMock,
1477
-
return_value=(existing_list_record, None),
1466
+
return_value=existing_uris,
1478
1467
),
1479
1468
patch(
1480
1469
"backend.api.albums.mutations.upsert_album_list_record",
+4
-5
backend/tests/api/test_playlist_by_uri.py
+4
-5
backend/tests/api/test_playlist_by_uri.py
···
85
85
test_app: FastAPI, db_session: AsyncSession, test_playlist: Playlist
86
86
) -> None:
87
87
"""lookup existing playlist by AT-URI returns 200."""
88
-
mock_record = {"value": {"items": []}}
89
88
with patch(
90
-
"backend.api.lists.playlists.get_record_public_resilient",
89
+
"backend.api.lists.playlists.fetch_list_item_uris",
91
90
new_callable=AsyncMock,
92
-
return_value=(mock_record, None),
91
+
return_value=[],
93
92
):
94
93
async with AsyncClient(
95
94
transport=ASGITransport(app=test_app), base_url="http://test"
···
128
127
) -> None:
129
128
"""playlist exists in DB but PDS record is gone — returns 404 not 500."""
130
129
with patch(
131
-
"backend.api.lists.playlists.get_record_public_resilient",
130
+
"backend.api.lists.playlists.fetch_list_item_uris",
132
131
new_callable=AsyncMock,
133
132
side_effect=RecordNotFound("record not found"),
134
133
):
···
149
148
) -> None:
150
149
"""playlist exists in DB but PDS record is gone — returns 404 not 500."""
151
150
with patch(
152
-
"backend.api.lists.playlists.get_record_public_resilient",
151
+
"backend.api.lists.playlists.fetch_list_item_uris",
153
152
new_callable=AsyncMock,
154
153
side_effect=RecordNotFound("record not found"),
155
154
):
+8
-32
backend/tests/api/test_playlist_liked_state.py
+8
-32
backend/tests/api/test_playlist_liked_state.py
···
144
144
this is a regression test for the bug where playlist tracks never showed
145
145
the liked state even when the user had liked them.
146
146
"""
147
-
# mock the ATProto record fetch to return our test tracks
148
-
mock_record_data = {
149
-
"value": {
150
-
"items": [
151
-
{
152
-
"subject": {
153
-
"uri": track.atproto_record_uri,
154
-
"cid": track.atproto_record_cid,
155
-
}
156
-
}
157
-
for track in test_tracks
158
-
]
159
-
}
160
-
}
147
+
# mock the ATProto list fetch to return our test track URIs
148
+
mock_track_uris = [track.atproto_record_uri for track in test_tracks]
161
149
162
150
# override get_optional_session to return our test user session
163
151
mock_session = MockSession()
···
168
156
test_app.dependency_overrides[get_optional_session] = _override_session
169
157
170
158
with patch(
171
-
"backend.api.lists.playlists.get_record_public_resilient",
159
+
"backend.api.lists.playlists.fetch_list_item_uris",
172
160
new_callable=AsyncMock,
173
-
return_value=(mock_record_data, None),
161
+
return_value=mock_track_uris,
174
162
):
175
163
async with AsyncClient(
176
164
transport=ASGITransport(app=test_app),
···
209
197
210
198
even if tracks have likes, unauthenticated users should not see their own liked state.
211
199
"""
212
-
# mock the ATProto record fetch to return our test tracks
213
-
mock_record_data = {
214
-
"value": {
215
-
"items": [
216
-
{
217
-
"subject": {
218
-
"uri": track.atproto_record_uri,
219
-
"cid": track.atproto_record_cid,
220
-
}
221
-
}
222
-
for track in test_tracks
223
-
]
224
-
}
225
-
}
200
+
# mock the ATProto list fetch to return our test track URIs
201
+
mock_track_uris = [track.atproto_record_uri for track in test_tracks]
226
202
227
203
# override get_optional_session to return None (unauthenticated)
228
204
async def _override_no_session() -> None:
···
231
207
test_app.dependency_overrides[get_optional_session] = _override_no_session
232
208
233
209
with patch(
234
-
"backend.api.lists.playlists.get_record_public_resilient",
210
+
"backend.api.lists.playlists.fetch_list_item_uris",
235
211
new_callable=AsyncMock,
236
-
return_value=(mock_record_data, None),
212
+
return_value=mock_track_uris,
237
213
):
238
214
async with AsyncClient(
239
215
transport=ASGITransport(app=test_app),