Firehose
The firehose is the whole network’s event stream, delivered as signed, verifiable CAR blocks over a websocket. For the prose, see Firehose; if you do not need cryptographic verifiability, Jetstream is cheaper to consume.
Subscribe to repository events
The minimum: connect, and print every frame.
from atproto import FirehoseSubscribeReposClient, firehose_models, parse_subscribe_repos_message
client = FirehoseSubscribeReposClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
client.start(on_message_handler)
import asyncio
from atproto import (
AsyncFirehoseSubscribeReposClient,
firehose_models,
parse_subscribe_repos_message,
)
async def main() -> None:
client = AsyncFirehoseSubscribeReposClient()
async def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
await client.start(on_message_handler)
if __name__ == '__main__':
# use run() for a higher Python version
asyncio.get_event_loop().run_until_complete(main())
Subscribe to label events
A separate stream, carrying moderation labels rather than repository commits.
from atproto import FirehoseSubscribeLabelsClient, firehose_models, models, parse_subscribe_labels_message
client = FirehoseSubscribeLabelsClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
labels_message = parse_subscribe_labels_message(message)
if not isinstance(labels_message, models.ComAtprotoLabelSubscribeLabels.Labels):
return
for label in labels_message.labels:
neg = '(NEG)' if label.neg else ''
print(f'[{label.cts}] ({label.src}) {label.uri} => {label.val} {neg}')
client.start(on_message_handler)
Decode commits into records
A commit frame carries a CAR file of blocks. Getting the records out means decoding it and looking up each operation’s CID.
import multiprocessing
import signal
import sys
import time
from collections import defaultdict
from types import FrameType
from typing import Any
from atproto import CAR, AtUri, FirehoseSubscribeReposClient, firehose_models, models, parse_subscribe_repos_message
_INTERESTED_RECORDS = {
models.ids.AppBskyFeedLike: models.AppBskyFeedLike,
models.ids.AppBskyFeedPost: models.AppBskyFeedPost,
models.ids.AppBskyGraphFollow: models.AppBskyGraphFollow,
}
def _get_ops_by_type(commit: models.ComAtprotoSyncSubscribeRepos.Commit) -> defaultdict:
operation_by_type = defaultdict(lambda: {'created': [], 'deleted': []})
car = CAR.from_bytes(commit.blocks)
for op in commit.ops:
if op.action == 'update':
# not supported yet
continue
uri = AtUri.from_str(f'at://{commit.repo}/{op.path}')
if op.action == 'create':
if not op.cid:
continue
create_info = {'uri': str(uri), 'cid': str(op.cid), 'author': commit.repo}
record_raw_data = car.blocks.get(op.cid)
if not record_raw_data:
continue
record = models.get_or_create(record_raw_data, strict=False)
record_type = _INTERESTED_RECORDS.get(uri.collection)
if record_type and models.is_record_type(record, record_type):
operation_by_type[uri.collection]['created'].append({'record': record, **create_info})
if op.action == 'delete':
operation_by_type[uri.collection]['deleted'].append({'uri': str(uri)})
return operation_by_type
def worker_main(cursor_value: multiprocessing.Value, pool_queue: multiprocessing.Queue) -> None:
signal.signal(signal.SIGINT, signal.SIG_IGN) # we handle it in the main process
while True:
message = pool_queue.get()
commit = parse_subscribe_repos_message(message)
if not isinstance(commit, models.ComAtprotoSyncSubscribeRepos.Commit):
continue
if commit.seq % 20 == 0:
cursor_value.value = commit.seq
if not commit.blocks:
continue
ops = _get_ops_by_type(commit)
for created_post in ops[models.ids.AppBskyFeedPost]['created']:
author = created_post['author']
record = created_post['record']
inlined_text = record.text.replace('\n', ' ')
print(f'NEW POST [CREATED_AT={record.created_at}][AUTHOR={author}]: {inlined_text}')
def get_firehose_params(cursor_value: multiprocessing.Value) -> models.ComAtprotoSyncSubscribeRepos.Params:
return models.ComAtprotoSyncSubscribeRepos.Params(cursor=cursor_value.value)
def measure_events_per_second(func: callable) -> callable:
def wrapper(*args) -> Any:
wrapper.calls += 1
cur_time = time.time()
if cur_time - wrapper.start_time >= 1:
print(f'NETWORK LOAD: {wrapper.calls} events/second')
wrapper.start_time = cur_time
wrapper.calls = 0
return func(*args)
wrapper.calls = 0
wrapper.start_time = time.time()
return wrapper
def signal_handler(_: int, __: FrameType) -> None:
print('Keyboard interrupt received. Waiting for the queue to empty before terminating processes...')
# Stop receiving new messages
client.stop()
# Drain the messages queue
while not queue.empty():
print('Waiting for the queue to empty...')
time.sleep(0.2)
print('Queue is empty. Gracefully terminating processes...')
pool.terminate()
pool.join()
sys.exit(0)
if __name__ == '__main__':
signal.signal(signal.SIGINT, signal_handler)
start_cursor = None
params = None
cursor = multiprocessing.Value('i', 0)
if start_cursor is not None:
cursor = multiprocessing.Value('i', start_cursor)
params = get_firehose_params(cursor)
client = FirehoseSubscribeReposClient(params)
workers_count = multiprocessing.cpu_count() * 2 - 1
max_queue_size = 10000
queue = multiprocessing.Queue(maxsize=max_queue_size)
pool = multiprocessing.Pool(workers_count, worker_main, (cursor, queue))
@measure_events_per_second
def on_message_handler(message: firehose_models.MessageFrame) -> None:
if cursor.value:
# we are using updating the cursor state here because of multiprocessing
# typically you can call client.update_params() directly on commit processing
client.update_params(get_firehose_params(cursor))
queue.put(message)
client.start(on_message_handler)
import asyncio
import signal
import time
from collections import defaultdict
from types import FrameType
from typing import Any
from atproto import (
CAR,
AsyncFirehoseSubscribeReposClient,
AtUri,
firehose_models,
models,
parse_subscribe_repos_message,
)
_INTERESTED_RECORDS = {
models.ids.AppBskyFeedLike: models.AppBskyFeedLike,
models.ids.AppBskyFeedPost: models.AppBskyFeedPost,
models.ids.AppBskyGraphFollow: models.AppBskyGraphFollow,
}
def _get_ops_by_type(commit: models.ComAtprotoSyncSubscribeRepos.Commit) -> defaultdict:
operation_by_type = defaultdict(lambda: {'created': [], 'deleted': []})
car = CAR.from_bytes(commit.blocks)
for op in commit.ops:
if op.action == 'update':
# not supported yet
continue
uri = AtUri.from_str(f'at://{commit.repo}/{op.path}')
if op.action == 'create':
if not op.cid:
continue
create_info = {'uri': str(uri), 'cid': str(op.cid), 'author': commit.repo}
record_raw_data = car.blocks.get(op.cid)
if not record_raw_data:
continue
record = models.get_or_create(record_raw_data, strict=False)
record_type = _INTERESTED_RECORDS.get(uri.collection)
if record_type and models.is_record_type(record, record_type):
operation_by_type[uri.collection]['created'].append({'record': record, **create_info})
if op.action == 'delete':
operation_by_type[uri.collection]['deleted'].append({'uri': str(uri)})
return operation_by_type
def measure_events_per_second(func: callable) -> callable:
def wrapper(*args) -> Any:
wrapper.calls += 1
cur_time = time.time()
if cur_time - wrapper.start_time >= 1:
print(f'NETWORK LOAD: {wrapper.calls} events/second')
wrapper.start_time = cur_time
wrapper.calls = 0
return func(*args)
wrapper.calls = 0
wrapper.start_time = time.time()
return wrapper
async def signal_handler(_: int, __: FrameType) -> None:
print('Keyboard interrupt received. Stopping...')
# Stop receiving new messages
await client.stop()
async def main(firehose_client: AsyncFirehoseSubscribeReposClient) -> None:
@measure_events_per_second
async def on_message_handler(message: firehose_models.MessageFrame) -> None:
commit = parse_subscribe_repos_message(message)
if not isinstance(commit, models.ComAtprotoSyncSubscribeRepos.Commit):
return
if commit.seq % 20 == 0:
firehose_client.update_params(models.ComAtprotoSyncSubscribeRepos.Params(cursor=commit.seq))
if not commit.blocks:
return
ops = _get_ops_by_type(commit)
for created_post in ops[models.ids.AppBskyFeedPost]['created']:
author = created_post['author']
record = created_post['record']
inlined_text = record.text.replace('\n', ' ')
print(f'NEW POST [CREATED_AT={record.created_at}][AUTHOR={author}]: {inlined_text}')
await client.start(on_message_handler)
if __name__ == '__main__':
signal.signal(signal.SIGINT, lambda _, __: asyncio.create_task(signal_handler(_, __)))
start_cursor = None
params = None
if start_cursor is not None:
params = models.ComAtprotoSyncSubscribeRepos.Params(cursor=start_cursor)
client = AsyncFirehoseSubscribeReposClient(params)
# use run() for a higher Python version
asyncio.get_event_loop().run_until_complete(main(client))
Stop the client
stop() closes the connection after the current message.
import threading
import time
from atproto import FirehoseSubscribeReposClient, firehose_models, parse_subscribe_repos_message
_STOP_AFTER_SECONDS = 3
client = FirehoseSubscribeReposClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
def _stop_after_n_sec() -> None:
time.sleep(_STOP_AFTER_SECONDS)
client.stop()
# run our sleep functions in another thread
threading.Thread(target=_stop_after_n_sec).start()
# run the client for N seconds
client.start(on_message_handler)
print(f'Successfully stopped after {_STOP_AFTER_SECONDS} seconds!')
import asyncio
from atproto import (
AsyncFirehoseSubscribeReposClient,
firehose_models,
parse_subscribe_repos_message,
)
_STOP_AFTER_SECONDS = 3
async def main() -> None:
client = AsyncFirehoseSubscribeReposClient()
async def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
async def _stop_after_n_sec() -> None:
await asyncio.sleep(_STOP_AFTER_SECONDS)
await client.stop()
# save ref to task to eliminate problems with GC
_stop_after_n_sec_task = asyncio.create_task(_stop_after_n_sec())
await client.start(on_message_handler)
await _stop_after_n_sec_task
print(f'Successfully stopped after {_STOP_AFTER_SECONDS} seconds!')
if __name__ == '__main__':
# use run() for a higher Python version
asyncio.get_event_loop().run_until_complete(main())
Handle errors
A long-lived stream will disconnect. Catch the error and reconnect from your stored cursor.
from atproto import FirehoseSubscribeReposClient, firehose_models, parse_subscribe_repos_message
client = FirehoseSubscribeReposClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
raise ValueError('Failed to process message')
def on_callback_error_handler(error: BaseException) -> None:
print('Got error!', error)
client.start(on_message_handler, on_callback_error_handler)