Skip to content

Stream data

nebra.stream(collections=(), dids=(), handles=(), message_handler=print, cursor=0, base_url=None, geo='us-west', compress=True, kinds=('commit',))

Programmatic interface for streaming Jetstream messages.

This function streams Jetstream messages and passes them to the provided message_handler. It can be called directly from Python code.

Parameters:

Name Type Description Default
collections Sequence[str]

The collections to subscribe to. Defaults to empty tuple.

 ()
dids Sequence[str]

The DIDs to subscribe to. Defaults to empty tuple.

 ()
handles Sequence[str]

The ATProto handles to subscribe to. Defaults to empty tuple.

 ()
message_handler Callable[[str], None]

Function to handle incoming messages. Defaults to print.

 print
cursor int

The cursor to start from. Defaults to 0.

 0
base_url str

The Jetstream URL to connect to. Defaults to None.

 None
geo (us - west, us - east)

The geography to use for public Jetstream. Defaults to "us-west".

 "us-west"
compress bool

Whether to enable compression. Defaults to True.

 True
kinds Sequence[str]

The kinds of events to subscribe to. Defaults to ("commit",).

 ('commit',)
Source code in nebra/jetstream.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def stream(
    collections: t.Sequence[str] = (),
    dids: t.Sequence[str] = (),
    handles: t.Sequence[str] = (),
    message_handler: Callable = print,
    cursor: int = 0,
    base_url: str | None = None,
    geo: t.Literal["us-west", "us-east"] = "us-west",
    compress: bool = True,
    kinds: t.Sequence[str] = ("commit",),
) -> None:
    """Programmatic interface for streaming Jetstream messages.

    This function streams Jetstream messages and passes them to the provided
    `message_handler`. It can be called directly from Python code.

    Parameters
    ----------
    collections : Sequence[str], optional
        The collections to subscribe to. Defaults to empty tuple.
    dids : Sequence[str], optional
        The DIDs to subscribe to. Defaults to empty tuple.
    handles : Sequence[str], optional
        The ATProto handles to subscribe to. Defaults to empty tuple.
    message_handler : Callable[[str], None], optional
        Function to handle incoming messages. Defaults to print.
    cursor : int, optional
        The cursor to start from. Defaults to 0.
    base_url : str, optional
        The Jetstream URL to connect to. Defaults to None.
    geo : {"us-west", "us-east"}, optional
        The geography to use for public Jetstream. Defaults to "us-west".
    compress : bool, optional
        Whether to enable compression. Defaults to True.
    kinds : Sequence[str], optional
        The kinds of events to subscribe to. Defaults to ("commit",).
    """
    _run_stream(
        collections=collections,
        dids=dids,
        handles=handles,
        message_handler=message_handler,
        cursor=cursor,
        base_url=base_url,
        geo=geo,
        compress=compress,
        kinds=kinds,
    )