Something went wrong. Try again.
Retro Bulletin Board Systems on atproto. Web app and TUI.
lazy mirror of [alyraffauf/atbbs](https://github.com/alyraffauf/atbbs) atbbs.xyz
forums python tui atproto bbs
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104import asyncio
import httpx
from core.models import ( AtUri, BBS, Board, Post, Site, BBSNotFoundError, NoBBSError, NetworkError, make_at_uri,)from core import lexiconfrom core.cache import TTLCachefrom core.constellation import get_root_postsfrom core.records import list_pds_records, post_from_recordfrom core.slingshot import get_record, get_records_batch, resolve_identity
_bbs_cache = TTLCache(ttl_seconds=300) # 5 minutes
def invalidate_bbs_cache(): _bbs_cache.clear()
async def resolve_bbs(client: httpx.AsyncClient, handle: str) -> BBS: cached = _bbs_cache.get(handle) if cached: return cached bbs = await _resolve_bbs(client, handle) _bbs_cache.set(handle, bbs) return bbs
async def _resolve_bbs(client: httpx.AsyncClient, handle: str) -> BBS: """Handle -> fully resolved BBS config.""" try: identity = await resolve_identity(client, handle) except httpx.HTTPStatusError: raise BBSNotFoundError(f"Could not resolve handle: {handle}") except httpx.TransportError: raise NetworkError("Could not reach the network.")
try: site_record = await get_record(client, identity.did, lexicon.SITE, "self") except httpx.HTTPStatusError: raise NoBBSError(f"{handle} isn't running a BBS.") except httpx.TransportError: raise NetworkError("Could not reach the network.")
site_value = site_record.value site_uri = make_at_uri(identity.did, lexicon.SITE, "self")
# Fetch boards and news concurrently board_uris = site_value["boards"] board_tasks = [] for uri in board_uris: parsed = AtUri.parse(uri) board_tasks.append( get_record(client, parsed.did, parsed.collection, parsed.rkey) ) news_task = get_root_posts(client, site_uri, did=identity.did)
results = await asyncio.gather(*board_tasks, news_task, return_exceptions=True) board_records = results[: len(board_uris)] news_result = results[len(board_uris)]
boards = [] for uri, record in zip(board_uris, board_records): if isinstance(record, BaseException): continue parsed = AtUri.parse(uri) boards.append( Board( slug=parsed.rkey, name=record.value["name"], description=record.value["description"], created_at=record.value["createdAt"], updated_at=record.value.get("updatedAt"), ) )
# Hydrate news posts (only from the sysop's repo — server-side filtered) if isinstance(news_result, BaseException): news_records = [] else: news_records = await get_records_batch(client, news_result.records) news = [post_from_record(record, identity) for record in news_records] news.sort(key=lambda post: post.created_at, reverse=True)
site = Site( name=site_value["name"], description=site_value["description"], intro=site_value["intro"], boards=boards, created_at=site_value.get("createdAt", ""), updated_at=site_value.get("updatedAt"), )
return BBS(identity=identity, site=site, news=news)