Broker API#

The Broker API is for provider plugins and user-facing commands that need to inspect or manage broker state.

These methods do not start the broker:

Broker.current()
Broker.running()
Broker.status()
Broker.service(NAME).check()
Broker.service(NAME).status()
Broker.service(NAME).running()
Broker.service(NAME).ready()
Broker.service(NAME).endpoint(ready=True)
Broker.events()
Broker.emit_event()

Startup methods are explicit:

Broker.start()
Broker.start_services()
Broker.service(NAME).start()
Broker.service(NAME).wait(start=True)
Broker.restart()
Broker.service(NAME).restart()

Broker.start() starts the broker process and any services enabled for broker startup. Broker.start_services() and Broker.service(NAME).start() start named services explicitly. Broker.start_services() and Broker.restart_services() require at least one service name; they never interpret an empty list as “all services.”

A multi-service start preflights service names, dependencies, and runtime availability. If a later process launch still fails, the supervisor stops processes and dependencies newly launched by that call while leaving pre-existing services alone.

Broker.service(NAME).wait(start=False) waits on an already running broker and does not start a process by itself.

Context managers are explicit lifecycle helpers:

from conda_broker import Broker

with Broker.current().started() as broker:
    print(broker.status().to_dict())

from conda_broker import Broker

with Broker.current().service("package-cache").started(wait=True) as service:
    endpoint = service.endpoint(ready=True)

started() context managers clean up only what they started. If the broker or service was already running before entering the with block, it is left running on exit. service.started(wait=True) raises ServiceNotReadyError if readiness is not reached, then cleans up anything the context started.

Plugin status commands can render a compact service report without starting anything:

from conda_broker import Broker

check = Broker.current().service("package-cache").check()
print(check.to_dict())

Service-handle queries take a fast path when the broker is stopped: they do not import provider entry points just to answer a runtime question. Service.check() then reports reason="broker-unavailable", while Service.status(), running(), ready(), and endpoint() return an absent or false result. Broker.status() is the strict, discovery-aware query.

Broker.emit_event() and Service.emit_event() return a typed ServiceEvent. They append directly to local state when no broker is running and retry that local path if the broker disappears during IPC.

For plugin-owned conda my-plugin services start|stop|status commands, use the Plugin Command API.

Broker object API for conda-broker users and provider plugins.

class conda_broker.api.Broker(paths: ServicePaths = <factory>)[source]#

Client for the current user’s conda-broker process.

classmethod current(paths: ServicePaths | None = None) → Broker[source]#: Return a broker handle for the current conda-broker user context.

emit_event(event_type: str, *, service: str | None = None, message: str = '', data: dict[str, Any] | None = None) → ServiceEvent[source]#: Record a provider event without forcing broker startup.

events(*, limit: int | None = None) → dict[str, Any][source]#: Read broker and service events without starting the broker.

list_services() → dict[str, Any][source]#: List discovered service definitions.

restart(*, timeout_s: float = 5.0) → BrokerState[source]#: Restart the broker process.

restart_services(services: str | list[str] | tuple[str, ...], *, timeout_s: float = 5.0) → StatusSnapshot[source]#: Restart selected services, starting the broker first if needed.

running() → bool[source]#: Return whether the broker is reachable without starting it.

service(name: str) → Service[source]#: Return a lightweight handle for one service name.

set_enabled(services: str | list[str] | tuple[str, ...], enabled: bool) → dict[str, Any][source]#: Enable or disable services for broker startup.

start(*, timeout_s: float = 5.0) → BrokerState[source]#: Start the broker if needed and wait until it accepts IPC.

start_services(services: str | list[str] | tuple[str, ...], *, timeout_s: float = 5.0) → StatusSnapshot[source]#: Start selected services, starting the broker first if needed.

started(*, timeout_s: float = 5.0, stop_timeout_s: float = 5.0) → BrokerContext[source]#

Return a context manager that keeps the broker running.

The context manager stops the broker on exit only when it started the broker on entry.

status(service: str | None = None) → StatusSnapshot[source]#: Return broker and service status without starting the broker.

stop(*, timeout_s: float = 5.0) → dict[str, Any][source]#: Stop the broker process and wait for lifecycle ownership to release.

stop_services(services: str | list[str] | tuple[str, ...]) → StatusSnapshot[source]#: Stop selected services without stopping the broker.

wait(service: str, *, timeout_s: float = 30.0, start_service: bool = False) → StatusSnapshot[source]#

Wait for one service to become ready.

The broker is only started when start_service is true.

class conda_broker.api.BrokerContext(broker: Broker, timeout_s: float = 5.0, stop_timeout_s: float = 5.0, _started_broker: bool = False)[source]#: Context manager returned by Broker.started().

class conda_broker.api.BrokerState(running: bool, started: bool | None = None, raw: dict[str, Any]=<factory>)[source]#: Reachability state for the local broker process.

class conda_broker.api.Service(broker: Broker, name: str)[source]#

Handle for one broker-managed service.

check(endpoint: str = 'default') → ServiceCheck[source]#

Return a compact service report without starting the broker.

This is intended for plugin status or doctor commands that want stable JSON and human output without reimplementing broker-state logic.

disable() → dict[str, Any][source]#: Disable this service for broker startup.

emit_event(event_type: str, *, message: str = '', data: dict[str, Any] | None = None) → ServiceEvent[source]#: Record an event for this service without forcing broker startup.

enable() → dict[str, Any][source]#: Enable this service for broker startup.

endpoint(name: str = 'default', *, ready: bool = False) → EndpointStatus | None[source]#

Return a resolved endpoint.

When ready is true, this returns None unless the service is already ready and the selected endpoint has a resolved URL.

ready() → bool[source]#: Return whether the service is ready for use.

restart(*, timeout_s: float = 5.0) → StatusSnapshot[source]#: Restart this service, starting the broker first if needed.

running() → bool[source]#: Return whether the service process is running.

start(*, timeout_s: float = 5.0) → StatusSnapshot[source]#: Start this service, starting the broker first if needed.

started(*, timeout_s: float = 5.0, wait: bool = False, wait_timeout_s: float = 30.0, stop_timeout_s: float = 5.0) → ServiceContext[source]#

Return a context manager that keeps this service running.

The context manager stops the service on exit only when it started the service on entry. If starting the service also started the broker, the broker is stopped on exit as well.

status() → ServiceStatus | None[source]#

Return service status, or None when unavailable.

This query never starts the broker and is safe for opportunistic plugin fast paths.

stop() → StatusSnapshot[source]#: Stop this service.

wait(*, timeout_s: float = 30.0, start: bool = False) → StatusSnapshot[source]#: Wait for this service to become ready.

class conda_broker.api.ServiceCheck(name: str, available: bool, running: bool = False, ready: bool = False, enabled: bool = False, state: str = 'unknown', health: str = 'unknown', endpoint: EndpointStatus | None = None, reason: str | None = None, status: ServiceStatus | None = None)[source]#: Compact status report for plugin CLIs and optional integrations.

class conda_broker.api.ServiceContext(service: Service, timeout_s: float = 5.0, wait: bool = False, wait_timeout_s: float = 30.0, stop_timeout_s: float = 5.0, _started_broker: bool = False, _started_service: bool = False)[source]#

Context manager returned by Service.started().

close() → None[source]#: Release lifecycle ownership acquired on entry.

class conda_broker.api.StatusSnapshot(broker: BrokerState | None = None, services: tuple[ServiceStatus, ...]=(), raw: dict[str, Any]=<factory>)[source]#: Broker and service status returned by API queries.