soulfire.session.BotSession¶

class soulfire.session.BotSession(stream: BotEventStreamFactory, options: BotSessionOptions)[source]¶

classmethod open(stream: BotEventStreamFactory, options: BotSessionOptions | None = None, *, ready_timeout: float | None = None) → effect_py.Effect[BotSession, soulfire.errors.SoulFireOperationError, effect_py.Scope][source]¶

property state: BotSessionState¶

events() → soulfire.streams.Stream[soulfire.bot_live_pb2.BotEvent, soulfire.errors.SoulFireOperationError][source]¶

wait_for(predicate: collections.abc.Callable[[soulfire.bot_live_pb2.BotEvent, BotSessionState], bool], *, timeout: float | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotEvent, soulfire.errors.SoulFireOperationError][source]¶

wait_for_state(predicate: collections.abc.Callable[[BotSessionState], bool]) → effect_py.Effect[BotSessionState, soulfire.errors.SoulFireOperationError][source]¶

once(event_name: str, *, timeout: float | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotEvent, soulfire.errors.SoulFireOperationError][source]¶

close() → effect_py.Effect[None][source]¶