soulfire.fleet.SoulFireFleet

class soulfire.fleet.SoulFireFleet(instance: soulfire.client.SoulFireInstance, capabilities: soulfire.connection.CapabilitySet | None)

select(selector: FleetSelector | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[tuple[FleetBot, ...], soulfire.errors.SoulFireOperationError]

start(selector: FleetSelector | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.bot_pb2.BotStatus], soulfire.errors.SoulFireOperationError]

stop(selector: FleetSelector | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.bot_pb2.BotStatus], soulfire.errors.SoulFireOperationError]

restart(selector: FleetSelector | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.bot_pb2.BotStatus], soulfire.errors.SoulFireOperationError]

start_tasks[ResultT: google.protobuf.message.Message](selector: FleetSelector, task_input: google.protobuf.message.Message | collections.abc.Callable[[FleetBot, int, int], google.protobuf.message.Message | effect_py.Effect[google.protobuf.message.Message, soulfire.errors.SoulFireOperationError]], result_type: type[ResultT], *, options: FleetTaskStartOptions | None = None) → effect_py.Effect[SoulFireFleetTaskGroup[ResultT], soulfire.errors.SoulFireOperationError]

distribute[ItemT](items: collections.abc.Iterable[ItemT], selector: FleetSelector | None = None, *, strategy: Literal['round-robin', 'contiguous'] = 'round-robin', maximum_items_per_bot: int | None = None, require_all: bool = True, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[tuple[FleetAssignment[ItemT], ...], soulfire.errors.SoulFireOperationError]