/usr/lib/python3/dist-packages/bioblend/galaxy/objects/galaxy_instance.py is in python3-bioblend 0.7.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """
A representation of a Galaxy instance based on oo wrappers.
"""
import time
import bioblend
import bioblend.galaxy
from . import client
# dataset states corresponding to a 'pending' condition
_PENDING_DS_STATES = set(
["new", "upload", "queued", "running", "setting_metadata"]
)
def _get_error_info(hda):
msg = hda.id
try:
msg += ' (%s): ' % hda.name
msg += hda.wrapped['misc_info']
except Exception: # avoid 'error while generating an error report'
msg += ': error'
return msg
class GalaxyInstance(object):
"""
A representation of an instance of Galaxy, identified by a URL and
a user's API key.
:type url: str
:param url: a FQDN or IP for a given instance of Galaxy. For example:
``http://127.0.0.1:8080``
:type api_key: str
:param api_key: user's API key for the given instance of Galaxy, obtained
from the Galaxy web UI.
This is actually a factory class which instantiates the entity-specific
clients.
Example: get a list of all histories for a user with API key 'foo'::
from bioblend.galaxy.objects import *
gi = GalaxyInstance('http://127.0.0.1:8080', 'foo')
histories = gi.histories.list()
"""
def __init__(self, url, api_key=None, email=None, password=None):
self.gi = bioblend.galaxy.GalaxyInstance(url, api_key, email, password)
self.log = bioblend.log
self.__histories = client.ObjHistoryClient(self)
self.__libraries = client.ObjLibraryClient(self)
self.__workflows = client.ObjWorkflowClient(self)
self.__tools = client.ObjToolClient(self)
self.__jobs = client.ObjJobClient(self)
@property
def histories(self):
"""
Client module for Galaxy histories.
"""
return self.__histories
@property
def libraries(self):
"""
Client module for Galaxy libraries.
"""
return self.__libraries
@property
def workflows(self):
"""
Client module for Galaxy workflows.
"""
return self.__workflows
@property
def tools(self):
"""
Client module for Galaxy tools.
"""
return self.__tools
@property
def jobs(self):
"""
Client module for Galaxy jobs.
"""
return self.__jobs
def _wait_datasets(self, datasets, polling_interval, break_on_error=True):
"""
Wait for datasets to come out of the pending states.
:type datasets: :class:`~collections.Iterable` of
:class:`~.wrappers.Dataset`
:param datasets: datasets
:type polling_interval: float
:param polling_interval: polling interval in seconds
:type break_on_error: bool
:param break_on_error: if ``True``, raise a RuntimeError exception as
soon as at least one of the datasets is in the 'error' state.
.. warning::
This is a blocking operation that can take a very long time.
Also, note that this method does not return anything;
however, each input dataset is refreshed (possibly multiple
times) during the execution.
"""
def poll(ds_list):
pending = []
for ds in ds_list:
ds.refresh()
self.log.info('{0.id}: {0.state}'.format(ds))
if break_on_error and ds.state == 'error':
raise RuntimeError(_get_error_info(ds))
if ds.state in _PENDING_DS_STATES:
pending.append(ds)
return pending
self.log.info('waiting for datasets')
while datasets:
datasets = poll(datasets)
time.sleep(polling_interval)
|