planemo.galaxy.invocations package

Submodules

planemo.galaxy.invocations.api module

API interaction for Galaxy’s workflow invocation API.

Gives a mockable surface for testing, type contract consumed by Planemo, and builtin utilities around bioblend for working around transient request issues that have been observed in practice.

class planemo.galaxy.invocations.api.BioblendInvocationApi(ctx, user_gi)[source]

Bases: InvocationApi

get_invocation(invocation_id: str) → Invocation[source]
get_invocation_jobs(invocation_id: str, state: str | None = None) → List[Job][source]
get_invocation_summary(invocation_id: str, state: str | None = None) → InvocationJobsSummary[source]
get_job(job_id: str, full_details: bool = False) → Job[source]
class planemo.galaxy.invocations.api.Invocation[source]

Bases: TypedDict

history_id: str | None
id: str
state: str
steps: List[InvocationStep]
class planemo.galaxy.invocations.api.InvocationApi(*args, **kwargs)[source]

Bases: Protocol

get_invocation(invocation_id: str) → Invocation[source]
get_invocation_jobs(invocation_id: str, state: str | None = None) → List[Job][source]
get_invocation_summary(invocation_id: str, state: str | None = None) → InvocationJobsSummary[source]
get_job(job_id: str, full_details: bool = False) → Job[source]
class planemo.galaxy.invocations.api.InvocationJobsSummary[source]

Bases: TypedDict

states: Dict[str, int]
class planemo.galaxy.invocations.api.InvocationStep[source]

Bases: TypedDict

state: str | None
subworkflow_invocation_id: str | None
class planemo.galaxy.invocations.api.Job[source]

Bases: TypedDict

command_line: str | None
exit_code: int | None
history_id: str | None
id: str
state: str
stderr: str | None
stdout: str | None
tool_id: str | None
planemo.galaxy.invocations.api.invocation_state_terminal(state: str)[source]

planemo.galaxy.invocations.polling module

class planemo.galaxy.invocations.polling.PollingTracker(*args, **kwargs)[source]

Bases: Protocol

sleep() → None[source]
class planemo.galaxy.invocations.polling.PollingTrackerImpl(polling_backoff: int, timeout=None)[source]

Bases: PollingTracker

sleep()[source]
planemo.galaxy.invocations.polling.subworkflow_invocation_ids(invocation_api: InvocationApi, invocation_id: str) → List[str][source]
planemo.galaxy.invocations.polling.summary_job_state(job_states_summary: InvocationJobsSummary | None, fail_fast: bool = False)[source]
planemo.galaxy.invocations.polling.wait_for_invocation_and_jobs(ctx, invocation_id: str, invocation_api: InvocationApi, polling_tracker: PollingTracker, workflow_progress_display: WorkflowProgressDisplay, fail_fast: bool = False)[source]
planemo.galaxy.invocations.polling.workflow_in_error_message(ctx, invocation_id, last_exception, last_invocation, last_invocation_jobs, invocation_api: InvocationApi, workflow_progress_display: WorkflowProgressDisplay, fail_fast=False) → str | None[source]

Return an error message if workflow is in an error state.

planemo.galaxy.invocations.progress module

class planemo.galaxy.invocations.progress.Invocation[source]

Bases: TypedDict

id: str
state: str
steps: List[InvocationStep]
class planemo.galaxy.invocations.progress.InvocationJobsSummary[source]

Bases: TypedDict

states: Dict[str, int]
class planemo.galaxy.invocations.progress.InvocationStep[source]

Bases: TypedDict

id: str
state: str | None
subworkflow_invocation_id: str | None
class planemo.galaxy.invocations.progress.WorkflowProgress(display: DisplayConfiguration)[source]

Bases: Progress

add_bars()[source]
handle_invocation(invocation: Invocation, job_state_summary: InvocationJobsSummary)[source]
handle_subworkflow_counts(num: int, num_complete: int)[source]
property invocation_scheduling_terminal
invocation_state: str
job_count: int | None
jobs_color: str
jobs_completed: int | None
property jobs_terminal
num_errors: int
num_new: int
num_ok: int
num_paused: int
num_queued: int
num_running: int
num_subworkflows: int
num_subworkflows_complete: int
print_job_errors_once(ctx, invocation_api: InvocationApi, invocation_id: str, workflow_progress_display: WorkflowProgressDisplay)[source]

Print job errors only if they haven’t been printed before, tracking by job ID.

printed_job_errors: Set[str]
step_count: int | None
step_states: Dict[str, int]
steps_color: str
subworkflows_color: str
property terminal
class planemo.galaxy.invocations.progress.WorkflowProgressDisplay(invocation_id: str, display_configuration: DisplayConfiguration | None = None, galaxy_url: str | None = None)[source]

Bases: Live

all_subworkflows_complete()[source]
an_incomplete_subworkflow_id()[source]
handle_invocation(invocation: Invocation, job_state_summary: InvocationJobsSummary)[source]
handle_subworkflow_invocation(invocation: Invocation, job_state_summary: InvocationJobsSummary)[source]
new_steps: List[str]
subworkflow_invocation_id: str | None
subworkflow_invocation_ids_completed: Set[str]
subworkflow_invocation_ids_seen: Set[str]
planemo.galaxy.invocations.progress.as_group(*renderables)[source]
planemo.galaxy.invocations.progress.count_states(job_summary: InvocationJobsSummary | None, query_states: list[str]) → int[source]
planemo.galaxy.invocations.progress.error_count(job_summary: InvocationJobsSummary) → int[source]
planemo.galaxy.invocations.progress.job_count(job_summary: InvocationJobsSummary | None) → int[source]
planemo.galaxy.invocations.progress.ok_count(job_summary: InvocationJobsSummary) → int[source]
planemo.galaxy.invocations.progress.running_count(job_summary: InvocationJobsSummary) → int[source]
planemo.galaxy.invocations.progress.step_states(invocation: Invocation)[source]

planemo.galaxy.invocations.progress_display module

class planemo.galaxy.invocations.progress_display.DisplayConfiguration(*, include_nested_subworkflows: bool = True, include_job_state_breakdown: bool = True, subworkflows_as_panel: bool = True, divider: str = '◆', style_bar_back: str = 'bar.back', style_bar_complete: str = 'bar.complete', style_bar_finished: str = 'bar.finished', style_percent: str = 'progress.percentage', style_initializing: str = 'cyan', style_ok: str = 'green', style_running: str = 'green', style_error: str = 'red', style_header: str = 'bold', style_subworkflow_header: str = 'bold', label_header_prefix: str = 'Invocation ', label_subworkflow_header_prefix: str = 'Subworkflow Invocation ', label_progress_steps: str = 'Steps', label_progress_jobs: str = 'Jobs', label_progress_subworkflows: str = 'SubWFs', label_job_states_prefix: str = 'Job States', icon_state_ok: str = '🟢', icon_state_errors: str = '🔴', icon_state_new: str = '🆕', icon_state_queued: str = '⏳', icon_state_running: str = '👟', icon_state_paused: str = '⏸️')[source]

Bases: BaseModel

divider: str
icon_state_errors: str
icon_state_new: str
icon_state_ok: str
icon_state_paused: str
icon_state_queued: str
icon_state_running: str
include_job_state_breakdown: bool
include_nested_subworkflows: bool
label_header_prefix: str
label_job_states_prefix: str
label_progress_jobs: str
label_progress_steps: str
label_progress_subworkflows: str
label_subworkflow_header_prefix: str
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

style_bar_back: str
style_bar_complete: str
style_bar_finished: str
style_error: str
style_header: str
style_initializing: str
style_ok: str
style_percent: str
style_running: str
style_subworkflow_header: str
subworkflows_as_panel: bool

planemo.galaxy.invocations.simulations module

Simulate Galaxy workflows running on a server for testing purposes.

class planemo.galaxy.invocations.simulations.HasState(after: int, states: List[StateWithDuration])[source]

Bases: Ticks

final: str | None
property state
tick_when_active() → None[source]
class planemo.galaxy.invocations.simulations.Invocation(steps: List[InvocationStep], after: int, states: List[StateWithDuration])[source]

Bases: HasState

property active_steps: List[InvocationStep]
get_api_invocation() → Invocation[source]
get_api_jobs_summary() → InvocationJobsSummary[source]
get_invocation_by_id(invocation_id: str) → Invocation | None[source]
get_subworkflow_invocation(subworkflow_invocation_id: str) → Invocation[source]
get_subworkflow_invocation_by_step_index(index: int) → Invocation | None[source]
tick_when_active() → None[source]
class planemo.galaxy.invocations.simulations.InvocationStep(jobs: List[HasState], invocation: Invocation | None, after: int, states: List[StateWithDuration])[source]

Bases: HasState

property active_jobs: List[HasState]
invocation: Invocation | None
jobs: List[HasState] | None
tick_when_active() → None[source]
planemo.galaxy.invocations.simulations.Job

alias of HasState

class planemo.galaxy.invocations.simulations.StateWithDuration(state: str, duration: int)[source]

Bases: Ticks

tick_when_active() → None[source]
class planemo.galaxy.invocations.simulations.Ticks[source]

Bases: object

property active
after: int
tick() → None[source]
tick_when_active() → None[source]
planemo.galaxy.invocations.simulations.parse_after_from(simulation_object: dict) → int[source]
planemo.galaxy.invocations.simulations.parse_states_from(simulation_object: dict) → List[StateWithDuration][source]
planemo.galaxy.invocations.simulations.parse_workflow_simulation(workflow_simulation: dict) → Invocation[source]
planemo.galaxy.invocations.simulations.parse_workflow_simulation_from_string(workflow_simulation: str) → Invocation[source]
planemo.galaxy.invocations.simulations.parse_workflow_simulation_invocation(workflow_simulation_invocation: dict) → Invocation[source]
planemo.galaxy.invocations.simulations.parse_workflow_simulation_invocation_step(workflow_simulation_invocation_step: dict) → InvocationStep[source]
planemo.galaxy.invocations.simulations.parse_workflow_simulation_job(workflow_simulation_job: dict) → HasState[source]

Module contents