planemo.galaxy package

Subpackages

Submodules

planemo.galaxy.activity module

Module provides generic interface to running Galaxy tools and workflows.

planemo.galaxy.activity.execute(ctx: PlanemoCliContext, config: BaseGalaxyConfig, runnable: Runnable, job_path: str, fail_fast=False, **kwds) → RunResponse[source]: Execute a Galaxy activity.

planemo.galaxy.api module

A high-level interface to local Galaxy instances using bioblend.

planemo.galaxy.api.gi(port: int | None = None, url: str | None = None, key: str | None = None) → GalaxyInstance[source]: Return a bioblend GalaxyInstance for Galaxy on this port.

planemo.galaxy.api.user_api_key(admin_gi)[source]: Use an admin authenticated account to generate a user API key.

planemo.galaxy.config module

Abstractions for setting up a Galaxy instance.

planemo.galaxy.config.galaxy_config(ctx, runnables, **kwds)[source]: Set up a GalaxyConfig in an auto-cleaned context.

planemo.galaxy.distro_tools module

planemo.galaxy.ephemeris_sleep module

Utility to do a blocking sleep until a Galaxy instance is responsive. This is useful in docker images, in RUN steps, where one needs to wait for a currently starting Galaxy to be alive, before API requests can be made successfully. The script functions by making repeated requests to http(s)://fqdn/api/version, an API which requires no authentication to access.

class planemo.galaxy.ephemeris_sleep.SleepCondition[source]

Bases: object

cancel()[source]

planemo.galaxy.ephemeris_sleep.main()[source]: Main function

planemo.galaxy.ephemeris_sleep.sleep(galaxy_url, verbose=False, timeout=0, sleep_condition=None)[source]

planemo.galaxy.profiles module

This modules describes the abstraction of a Galaxy profile.

This is a workspace with a specific default configuration and shed tool setup. It is meant to be used with various serve commands.

planemo.galaxy.profiles.create_profile(ctx, profile_name, **kwds)[source]: Create a profile with the specified name.

planemo.galaxy.profiles.delete_profile(ctx, profile_name, **kwds)[source]: Delete profile with the specified name.

planemo.galaxy.profiles.ensure_profile(ctx, profile_name, **kwds)[source]: Ensure a Galaxy profile exists and return profile defaults.

planemo.galaxy.profiles.list_profiles(ctx, **kwds)[source]: Return a list of current profile names.

planemo.galaxy.run module

Utilities for calling Galaxy scripts.

planemo.galaxy.run.run_galaxy_command(ctx, command, env, action)[source]: Run Galaxy command with informative verbose logging.

planemo.galaxy.run.setup_venv(ctx, kwds: Dict[str, Any], config: LocalGalaxyConfig | None = None)[source]

planemo.galaxy.serve module

Abstractions for serving out development Galaxy servers.

planemo.galaxy.serve.serve(ctx, runnables=None, **kwds)[source]

planemo.galaxy.serve.serve_daemon(ctx, runnables=None, **kwds)[source]: Serve a daemonized Galaxy instance with artifacts defined by paths.

planemo.galaxy.upload_progress module

Progress display for Galaxy data uploads.

class planemo.galaxy.upload_progress.UploadJobsSummary(states: Dict[str, int] | None = None)[source]

Bases: object

Summary of upload job states.

class planemo.galaxy.upload_progress.UploadProgress(display: DisplayConfiguration)[source]

Bases: Progress

Progress tracker for upload jobs.

handle_upload_jobs(job_summary: UploadJobsSummary)[source]

Update progress based on upload job states.

Args:: job_summary: Summary of current job states

property jobs_terminal_count: int: Count of jobs in terminal states (completed, errors, paused).

print_job_errors_once(gi, upload_progress_display: UploadProgressDisplay)[source]

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

Args:: gi: GalaxyInstance for querying job details upload_progress_display: Display instance for console output

property terminal: bool: Check if all upload jobs are in terminal state.

class planemo.galaxy.upload_progress.UploadProgressDisplay(history_id: str, display_configuration: DisplayConfiguration | None = None, galaxy_url: str | None = None)[source]

Bases: Live

Live display for upload progress with Rich panel.

get_history_ui_link() → str | None[source]

Get the Galaxy UI link for the history.

Returns:: URL to the history in Galaxy UI, or None if galaxy_url not set

update_jobs(upload_jobs: List[Dict[str, Any]], job_summary: UploadJobsSummary)[source]

Update progress display with current upload job states.

Args:: upload_jobs: List of upload job dictionaries job_summary: Summary of job states

planemo.galaxy.upload_progress.count_states(job_summary: UploadJobsSummary | None, query_states: List[str]) → int[source]: Count jobs in specific states.

planemo.galaxy.upload_progress.error_count(job_summary: UploadJobsSummary) → int[source]: Count jobs in error states.

planemo.galaxy.upload_progress.job_count(job_summary: UploadJobsSummary | None) → int[source]: Count total jobs across all states.

planemo.galaxy.upload_progress.ok_count(job_summary: UploadJobsSummary) → int[source]: Count successfully completed jobs.

planemo.galaxy.workflows module

Utilities for Galaxy workflows.

planemo.galaxy.workflows.describe_outputs(runnable, gi=None)[source]: Return a list of WorkflowOutput objects for target workflow.

planemo.galaxy.workflows.fetch_workflow_from_trs(trs_id: str) → Dict[str, Any][source]

Fetch a workflow definition directly from a TRS endpoint.

Args:

trs_id: TRS ID in format: [#]workflow/github.com/org/repo/workflow_name[/version]: or a full TRS URL

Returns:

Workflow definition dict (gxformat2 or GA format)

Raises:

ValueError: If the TRS ID is invalid or workflow cannot be fetched

planemo.galaxy.workflows.import_workflow(path, user_gi, from_path=False)[source]: Import a workflow path to specified Galaxy instance.

planemo.galaxy.workflows.import_workflow_from_trs(trs_uri: str, importer: TrsImporter) → Dict[str, Any][source]

Import a workflow from a TRS endpoint.

Args:: trs_uri: TRS URI in format trs://[#]workflow/github.com/org/repo/workflow_name[/version] importer: TrsImporter implementation to use for importing
Returns:: Workflow dict with ‘id’ and other metadata

planemo.galaxy.workflows.is_trs_identifier(identifier: str) → bool[source]

Check if the identifier is a TRS ID or TRS URI.

Args:: identifier: The workflow identifier to check
Returns:: True if it’s a TRS ID or URI, False otherwise

Module contents

Entry-point for Galaxy specific functionality in Planemo.

planemo.galaxy.galaxy_config(ctx, runnables, **kwds)[source]: Set up a GalaxyConfig in an auto-cleaned context.

planemo.galaxy.galaxy_serve(ctx, runnables=None, **kwds)

planemo.galaxy.run_galaxy_command(ctx, command, env, action)[source]: Run Galaxy command with informative verbose logging.

planemo.galaxy.setup_venv(ctx, kwds: Dict[str, Any], config: LocalGalaxyConfig | None = None)[source]