soulfire.PluginCatalog¶

class soulfire.PluginCatalog(client: soulfire.plugin_api_connect.PluginApiServiceClient, service_factory: ServiceFactory, plugins: tuple[soulfire.plugin_api_pb2.PluginApiDescriptor, ...] = (), reflective_client: connectrpc.client.ConnectClient | None = None)[source]¶

Bases: _PluginCatalogBase

require[T](module: SoulFirePluginModule[T]) → T[source]¶

service[T](client_type: collections.abc.Callable[..., T]) → T[source]¶

refresh(*, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → effect_py.Effect[tuple[soulfire.plugin_api_pb2.PluginApiDescriptor, ...], soulfire.errors.SoulFireOperationError][source]¶

descriptor_set(plugin_id: str, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → effect_py.Effect[bytes, soulfire.errors.SoulFireOperationError][source]¶

reflective(plugin_id: str, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → effect_py.Effect[ReflectivePlugin, soulfire.errors.SoulFireOperationError][source]¶

watch(*, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.plugin_api_pb2.PluginApiEvent, soulfire.errors.SoulFireOperationError][source]¶

events(*, plugin_ids: tuple[str, ...] = (), type_urls: tuple[str, ...] = (), instance_id: str | None = None, bot_id: str | None = None, task_id: str | None = None, after_sequence: int = 0, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.plugin_api_pb2.PluginEvent, soulfire.errors.SoulFireOperationError][source]¶

typed_events[T: google.protobuf.message.Message](plugin_id: str, message_type: type[T], *, instance_id: str | None = None, bot_id: str | None = None, task_id: str | None = None, after_sequence: int = 0, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → soulfire.streams.Stream[TypedPluginEvent[T], soulfire.errors.SoulFireOperationError][source]¶