aly.codes
Retro Bulletin Board Systems on atproto. Web app and TUI.
atbbs.xyz
python
tui
atproto
bbs
1import httpx
2
3from core import lexicon
4from core.models import BacklinkRef, BacklinksResponse
5
6BASE_URL = "https://constellation.microcosm.blue/xrpc"
7
8
9async def get_backlinks(
10 client: httpx.AsyncClient,
11 subject: str,
12 source: str,
13 limit: int = 50,
14 cursor: str | None = None,
15) -> BacklinksResponse:
16 """Query Constellation for records that link to a subject."""
17 params: dict[str, str | int] = {
18 "subject": subject,
19 "source": source,
20 "limit": limit,
21 }
22 if cursor is not None:
23 params["cursor"] = cursor
24 resp = await client.get(
25 f"{BASE_URL}/blue.microcosm.links.getBacklinks",
26 params=params,
27 )
28 resp.raise_for_status()
29 data = resp.json()
30 return BacklinksResponse(
31 total=data["total"],
32 records=[
33 BacklinkRef(did=r["did"], collection=r["collection"], rkey=r["rkey"])
34 for r in data["records"]
35 ],
36 cursor=data.get("cursor"),
37 )
38
39
40async def get_threads(
41 client: httpx.AsyncClient,
42 board_uri: str,
43 limit: int = 50,
44 cursor: str | None = None,
45) -> BacklinksResponse:
46 """Get threads pointing at a board."""
47 return await get_backlinks(
48 client,
49 subject=board_uri,
50 source=f"{lexicon.THREAD}:board",
51 limit=limit,
52 cursor=cursor,
53 )
54
55
56async def get_news(
57 client: httpx.AsyncClient,
58 site_uri: str,
59 limit: int = 50,
60 cursor: str | None = None,
61) -> BacklinksResponse:
62 """Get news posts pointing at a site."""
63 return await get_backlinks(
64 client,
65 subject=site_uri,
66 source=f"{lexicon.NEWS}:site",
67 limit=limit,
68 cursor=cursor,
69 )
70
71
72async def get_replies(
73 client: httpx.AsyncClient,
74 thread_uri: str,
75 limit: int = 50,
76 cursor: str | None = None,
77) -> BacklinksResponse:
78 """Get replies pointing at a thread."""
79 return await get_backlinks(
80 client,
81 subject=thread_uri,
82 source=f"{lexicon.REPLY}:subject",
83 limit=limit,
84 cursor=cursor,
85 )