Best method for monitoring a timeline for updates?

[quartzjer]

I've been playing around with simple cli-based clients, this is what I've come up with to monitor the timeline for new posts and it works but seems pretty inefficient as it is polling based. I've also tried wading through the bsky api docs but nothing is jumping out as an obvious and better alternative.

If this really is the right pattern, it might be nice to eventually have some of this logic bundled into the atproto client :)

import asyncio
import textwrap
from html import escape
import os
from dotenv import load_dotenv
from datetime import datetime
import humanize

from atproto import AsyncClient

load_dotenv()

FETCH_NOTIFICATIONS_DELAY_SEC = 5

def process_post(post, seen_posts):
    if post.cid not in seen_posts:
        author = post.author.display_name or post.author.handle
        created_at = post.indexed_at
        timestamp = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
        age = humanize.naturaltime(datetime.now().astimezone() - timestamp)

        text = textwrap.fill(post.record.text, width=70)
        text = escape(text)

        print(f"\n\033[94m[{created_at}] ({age}) \033[92m{author}\033[0m ({post.author.did}):\n")
        print(text)

        seen_posts.add(post.cid)
        return True
    return False

async def main() -> None:
    client = AsyncClient()
    await client.login(os.getenv('BSKY_HANDLE'), os.getenv('BSKY_APP_PASSWORD'))

    print("Monitoring timeline for new posts...")

    seen_posts = set()
    cursor = None
    oldest = None

    while True:
        try:
            timeline = await client.get_timeline(limit=1, cursor=cursor)
            print(f"Got {len(timeline.feed)} new posts")
            if not timeline.feed and not seen_posts:
                cursor = timeline.cursor # startup go till we get a post since muted posts count in the limit but aren't returned
                continue

            for fv in timeline.feed:
                if process_post(fv.post, seen_posts):
                    cursor = timeline.cursor
                else:
                    cursor = None
                at = datetime.fromisoformat(fv.post.indexed_at.replace('Z', '+00:00'))
                if oldest is None or at < oldest:
                    oldest = at
                    cursor = None # never page older
            
        except Exception as e:
            print(f"Error: {e}")
            cursor = None
            await asyncio.sleep(10)
            continue

        await asyncio.sleep(FETCH_NOTIFICATIONS_DELAY_SEC)

if __name__ == '__main__':
    asyncio.run(main())

[MarshalX]

Hi Jer! If the timeline is just a simple list of posts where the posts are filtered by a list of your followings and sorted by timestamp, then it could be more optimal to re-implement this logic and listen to firehose events of new posts. And do your filtering by following manually to populate your timeline

[quartzjer]

I started thinking down that path of using the firehose, but it got a bit more complicated faster than I was ready for. Right now the getTimeline nicely fully hydrates every post and automatically takes care of the mutes (though not without causing oddities w/ the limit parameter handling).

It would be ideal if the bsky app API's getTimeline endpoint accepted algo=chronological and a cursor, but I don't think it does (or I was doing it wrong).