Scheduler

The SchedulerDirector exposes coordinator objects that orchestrate several managers to satisfy Scheduler-specific workflows.

Usage Example

from gpp_client import GPPClient
from gpp_client import Director
client = GPPClient()
director = Director(client)
programs = await director.scheduler.program.get_all()

class gpp_client.directors.scheduler.SchedulerDirector(client: GPPClient)

Bases: BaseDirector

Facade for Scheduler-domain workflows.

The director instantiates and exposes coordinator objects that orchestrate multiple managers to fulfil complex Scheduler-specific tasks. Each coordinator receives the shared GPPClient instance injected into this director.

Parameters:

client (GPPClient) – The low-level API client used by all underlying managers.

program

Coordinates program data tailored to the Scheduler.

Type:

ProgramCoordinator

client: GPPClient

Authenticated low-level client reused by every manager / coordinator.

Coordinator Layer

Each coordinator bundles related manager calls into a single, domain-focused API. For example, ProgramCoordinator combines ProgramManager and ObservationManager to return a program plus its observation hierarchy in one asynchronous method.

class gpp_client.directors.scheduler.coordinators.program.ProgramCoordinator(client: GPPClient)

Bases: BaseCoordinator

Combines multiple managers to return views of a program and its observations. Includes subscriber to handle subscriptions related to programs.

client: GPPClient

Shared low-level client injected by the parent director.

async get_all(programs_list: list | None = None) → list[dict[str, Any]]

Fetch all programs with a complete group tree and observations.

Parameters:

programs_list (list, optional) – Optional filtering clause.

Returns:

A list of dictionaries representing the programs and their elements.

Return type:

list[dict[str, Any]]

async get_all_reference_labels(date: str | None = None) → list[tuple[str, str]]

Parameters:

date (str | None) – change the activeEnd parameter date. by default is the today date.

Returns:

return a list of tuples containing the program reference label and the id.

Return type:

list[tuple[str, str]]

Subscriber Integration

The ProgramCoordinator for the Scheduler Director currently includes a subscribe_to attribute that provides access to subscription functionality for testing purposes. This allows you to subscribe to real-time program edits and updates directly through the Scheduler Director.

Note

The subscriber is currently integrated into the Scheduler Director for testing. In future releases, the subscriber architecture may be restructured.

Usage Example

from gpp_client import GPPClient
from gpp_client import Director

client = GPPClient()
director = Director(client)

# Access the subscriber through the program coordinator
subscriber = director.scheduler.program.subscribe_to

# Subscribe to program edits
async for edit in await subscriber.get_edits(program_id="p-123"):
    print(f"Received edit: {edit}")

Available Methods

• get_edits(program_id: str | None): Subscribe to edits for a specific program. Returns an async iterator of program edit changes.

• get_calculations_updates(program_id: str | None): Subscribe to calculations updates for a specific program. Returns an async iterator of program calculations updates. This is only for te Observation Coordinator.