Jetstream
Jetstream delivers the same events as the firehose as plain JSON, filtered server-side, with no CAR or DAG-CBOR decoding. For the prose on filtering, cursors, compression and the metered archive, see Jetstream.
Subscribe to everything
from atproto import JetstreamClient, jetstream_models, models
client = JetstreamClient()
def on_message_handler(event: jetstream_models.SubscribeEventsMessage) -> None:
if isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Info):
# advisory about the stream itself; it carries no seq
print('info:', event.name, event.message)
return
print(event.seq, event.did, type(event).__name__)
client.start(on_message_handler)
Filter for posts
Filters are applied by the server, so you only pay for what you asked for.
from atproto import JetstreamClient, jetstream_models, models
client = JetstreamClient(params={'collections': [models.ids.AppBskyFeedPost], 'kinds': ['commit']})
def on_message_handler(event: jetstream_models.SubscribeEventsMessage) -> None:
if not isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
return
if event.operation != 'create':
return
# already decoded into a model by the client; a non-conforming record falls back to DotDict
print(f'[{event.seq}] {event.did}: {event.record.text}')
client.start(on_message_handler)
import asyncio
from atproto import AsyncJetstreamClient, jetstream_models, models
client = AsyncJetstreamClient(params={'collections': [models.ids.AppBskyFeedPost], 'kinds': ['commit']})
async def on_message_handler(event: jetstream_models.SubscribeEventsMessage) -> None:
if not isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
return
if event.operation != 'create':
return
# already decoded into a model by the client; a non-conforming record falls back to DotDict
print(f'[{event.seq}] {event.did}: {event.record.text}')
asyncio.run(client.start(on_message_handler))
Resume from a stored cursor
The client tracks the cursor across reconnects on its own. Persist it to survive a restart too.
"""Persist the cursor so a restart resumes where the previous run stopped."""
import os
import typing as t
from pathlib import Path
from atproto import JetstreamClient, jetstream_models, models
_CURSOR_FILE = Path('jetstream.cursor')
#: Saving on every event would hammer the disk
_SAVE_EVERY = 100
def load_cursor() -> t.Optional[int]:
try:
return int(_CURSOR_FILE.read_text())
except (OSError, ValueError):
# missing or truncated by a crash; start from the live tip
return None
def save_cursor(cursor: int) -> None:
# write to a temporary file and rename, so a crash cannot leave a half-written cursor
tmp_file = _CURSOR_FILE.with_suffix('.tmp')
tmp_file.write_text(str(cursor))
os.replace(tmp_file, _CURSOR_FILE)
params: models.NetworkBskyJetstreamSubscribeEvents.ParamsDict = {'kinds': ['commit']}
cursor = load_cursor()
if cursor is not None:
params['cursor'] = cursor
client = JetstreamClient(params=params)
processed = 0
def on_message_handler(event: jetstream_models.SubscribeEventsMessage) -> None:
global processed
if isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Info):
return
print(event.seq, event.did)
processed += 1
if processed % _SAVE_EVERY == 0 and client.cursor is not None:
save_cursor(client.cursor)
client.start(on_message_handler)
Replay the archive
snapshot() sweeps the sealed archive and stops. replay() sweeps it and continues into the live tail without a gap, so a consumer cannot tell where one ended and the other began.
Warning
The archive is metered in bytes downloaded, not requests, and the whole network is roughly 1.85 TB. Filter selectively and check bytes_downloaded. See Metering.
"""Replay one repository's whole history out of the Jetstream archive.
Needs an API key from https://bsky.network/account. The archive is metered in bytes, so
filter narrowly: this plan matches a single block.
"""
import os
from atproto import JetstreamClient, models
TARGET_DID = 'did:plc:kvwvcn5iqfooopmyzvb4qzba'
client = JetstreamClient(params={'dids': [TARGET_DID]}, api_key=os.environ['JETSTREAM_API_KEY'])
posts = 0
for event in client.snapshot(after_seq=0):
if not isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
continue
if event.collection != models.ids.AppBskyFeedPost or event.operation != 'create':
continue
posts += 1
print(f'[{event.seq}] {event.time} {event.record.text[:60]}')
print(f'\n{posts} posts, {client.bytes_downloaded:,} bytes downloaded')
"""Catch up on the archive, then keep streaming live, without a gap or a duplicate.
`replay()` sweeps the sealed archive first and cuts over to the live tail at the seam. It
never terminates.
The archive is metered in bytes, and how much a filter costs depends on how selective it is,
not on whether one is set. Every filter is sent to the planner, but segments carry per-DID
bloom filters: `dids` prunes hard, while a popular collection appears in nearly every block
and prunes almost nothing.
Measured against the whole archive, from `after_seq=0`:
dids=[one repo] -> 1 of 7,075 segments, 1 block
collections=[app.bsky.feed.post] -> 7,075 of 7,075 segments, 5,363,406 blocks
So this example filters to one repository. Its whole history is a single block, about
274 KB. To follow a busy collection instead, resume from a stored cursor rather than
sweeping from the beginning.
"""
import os
from atproto import JetstreamClient, models
TARGET_DID = 'did:plc:kvwvcn5iqfooopmyzvb4qzba'
client = JetstreamClient(params={'dids': [TARGET_DID]}, api_key=os.environ['JETSTREAM_API_KEY'])
# archived events arrive first, then the stream continues live from the seam
for event in client.replay(after_seq=0):
if not isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
continue
print(f'{event.seq} {event.collection} {event.operation}')
"""Async archive replay.
Downloads run on the event loop and decoding is offloaded, so the loop stays responsive
while blocks are being decoded.
"""
import asyncio
import os
from atproto import AsyncJetstreamClient, models
TARGET_DID = 'did:plc:kvwvcn5iqfooopmyzvb4qzba'
async def main() -> None:
client = AsyncJetstreamClient(params={'dids': [TARGET_DID]}, api_key=os.environ['JETSTREAM_API_KEY'])
collections: dict = {}
async for event in client.snapshot(after_seq=0):
if isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
collections[event.collection] = collections.get(event.collection, 0) + 1
for collection, count in sorted(collections.items(), key=lambda item: -item[1]):
print(f'{count:5} {collection}')
print(f'\n{client.bytes_downloaded:,} bytes downloaded')
asyncio.run(main())