Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self.d = {'a': 1, 'b': [2, 3], 'c': {'x': 4}}
self.assertRaises(TypeError, wrappers.Wrapper, self.d)
self.w = MockWrapper(self.d)
def upload_from_fs(lib, bnames, **kwargs):
tempdir = tempfile.mkdtemp(prefix='bioblend_test_')
try:
fnames = [os.path.join(tempdir, _) for _ in bnames]
for fn in fnames:
with open(fn, 'w') as f:
f.write(FOO_DATA)
dss = lib.upload_from_galaxy_fs(fnames, **kwargs)
finally:
shutil.rmtree(tempdir)
return dss, fnames
class MockWrapper(wrappers.Wrapper):
BASE_ATTRS = frozenset(['a', 'b'])
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@property
def gi_module(self):
return super().gi_module()
class TestWrapper(unittest.TestCase):
def setUp(self):
self.d = {'a': 1, 'b': [2, 3], 'c': {'x': 4}}
self.assertRaises(TypeError, wrappers.Wrapper, self.d)
self.w = MockWrapper(self.d)
Update this library dataset metadata. Some of the attributes that can be
modified are documented below.
:type name: str
:param name: Replace history dataset name with the given string
:type genome_build: str
:param genome_build: Replace history dataset genome build (dbkey)
"""
res = self.gi.gi.libraries.update_library_dataset(self.id, **kwds)
self.container.refresh()
self.__init__(res, self.container, gi=self.gi)
return self
class ContentInfo(Wrapper, metaclass=abc.ABCMeta):
"""
Instances of this class wrap dictionaries obtained by getting
``/api/{histories,libraries}//contents`` from Galaxy.
"""
BASE_ATTRS = Wrapper.BASE_ATTRS + ('type',)
@abc.abstractmethod
def __init__(self, info_dict, gi=None):
super().__init__(info_dict, gi=gi)
class LibraryContentInfo(ContentInfo):
"""
Instances of this class wrap dictionaries obtained by getting
``/api/libraries//contents`` from Galaxy.
"""
`_.
The value of an input dataset can also be a :class:`Dataset`
object, which will be automatically converted to the needed
format.
"""
for k, v in inputs.items():
if isinstance(v, Dataset):
inputs[k] = {'src': v.SRC, 'id': v.id}
out_dict = self.gi.gi.tools.run_tool(history.id, self.id, inputs)
outputs = [history.get_dataset(_['id']) for _ in out_dict['outputs']]
if wait:
self.gi._wait_datasets(outputs, polling_interval=polling_interval)
return outputs
class Job(Wrapper):
"""
Maps to a Galaxy job.
"""
BASE_ATTRS = ('id', 'state')
def __init__(self, j_dict, gi=None):
super().__init__(j_dict, gi=gi)
@property
def gi_module(self):
return self.gi.jobs
class Preview(Wrapper, metaclass=abc.ABCMeta):
"""
Abstract base class for Galaxy entity 'previews'.
"""
f_dict = self.gi.gi.libraries.show_folder(self.id, f_id)
return Folder(f_dict, self, gi=self.gi)
@property
def root_folder(self):
"""
The root folder of this library.
:rtype: :class:`~.Folder`
:return: the root folder of this library
"""
return self.get_folder(self.gi.gi.libraries._get_root_folder_id(self.id))
class Folder(Wrapper):
"""
Maps to a folder in a Galaxy library.
"""
BASE_ATTRS = Wrapper.BASE_ATTRS + ('description', 'deleted', 'item_count')
def __init__(self, f_dict, container, gi=None):
super().__init__(f_dict, gi=gi)
object.__setattr__(self, 'container', container)
@property
def parent(self):
"""
The parent folder of this folder. The parent of the root folder is
``None``.
:rtype: :class:`~.Folder`
def __init__(self, step_dict, parent):
super().__init__(step_dict, parent=parent, gi=parent.gi)
try:
stype = step_dict['type']
except KeyError:
raise ValueError('not a step dict')
if stype not in {'data_collection_input', 'data_input', 'parameter_input', 'pause', 'subworkflow', 'tool'}:
raise ValueError('Unknown step type: %r' % stype)
@property
def gi_module(self):
return self.gi.workflows
class Workflow(Wrapper):
"""
Workflows represent ordered sequences of computations on Galaxy.
A workflow defines a sequence of steps that produce one or more
results from an input dataset.
"""
BASE_ATTRS = Wrapper.BASE_ATTRS + (
'deleted', 'inputs', 'published', 'steps', 'tags'
)
POLLING_INTERVAL = 10 # for output state monitoring
def __init__(self, wf_dict, gi=None):
super().__init__(wf_dict, gi=gi)
missing_ids = []
if gi:
tools_list_by_id = [t.id for t in gi.tools.get_previews()]
class HistoryContentInfo(ContentInfo):
"""
Instances of this class wrap dictionaries obtained by getting
``/api/histories//contents`` from Galaxy.
"""
BASE_ATTRS = ContentInfo.BASE_ATTRS + ('deleted', 'state', 'visible')
def __init__(self, info_dict, gi=None):
super().__init__(info_dict, gi=gi)
@property
def gi_module(self):
return self.gi.histories
class DatasetContainer(Wrapper, metaclass=abc.ABCMeta):
"""
Abstract base class for dataset containers (histories and libraries).
"""
BASE_ATTRS = Wrapper.BASE_ATTRS + ('deleted',)
@abc.abstractmethod
def __init__(self, c_dict, content_infos=None, gi=None):
"""
:type content_infos: list of :class:`ContentInfo`
:param content_infos: info objects for the container's contents
"""
super().__init__(c_dict, gi=gi)
if content_infos is None:
content_infos = []
object.__setattr__(self, 'content_infos', content_infos)
Id of the folder container. Use :attr:`.container.id` instead.
"""
return self.container.id
def refresh(self):
"""
Re-fetch the attributes pertaining to this object.
Returns: self
"""
f_dict = self.gi.gi.libraries.show_folder(self.container.id, self.id)
self.__init__(f_dict, self.container, gi=self.gi)
return self
class Tool(Wrapper):
"""
Maps to a Galaxy tool.
"""
BASE_ATTRS = Wrapper.BASE_ATTRS + ('version',)
POLLING_INTERVAL = 10 # for output state monitoring
def __init__(self, t_dict, gi=None):
super().__init__(t_dict, gi=gi)
@property
def gi_module(self):
return self.gi.tools
def run(self, inputs, history, wait=False,
polling_interval=POLLING_INTERVAL):
"""
class Job(Wrapper):
"""
Maps to a Galaxy job.
"""
BASE_ATTRS = ('id', 'state')
def __init__(self, j_dict, gi=None):
super().__init__(j_dict, gi=gi)
@property
def gi_module(self):
return self.gi.jobs
class Preview(Wrapper, metaclass=abc.ABCMeta):
"""
Abstract base class for Galaxy entity 'previews'.
Classes derived from this one model the short summaries returned
by global getters such as ``/api/libraries``.
"""
BASE_ATTRS = Wrapper.BASE_ATTRS + ('deleted',)
@abc.abstractmethod
def __init__(self, pw_dict, gi=None):
super().__init__(pw_dict, gi=gi)
class LibraryPreview(Preview):
"""
Models Galaxy library 'previews'.
"""
return self.gi.gi.workflows.export_workflow_dict(self.id)
def delete(self):
"""
Delete this workflow.
.. warning::
Deleting a workflow is irreversible - all of the data from
the workflow will be permanently deleted.
"""
self.gi.workflows.delete(id_=self.id)
self.unmap()
class Dataset(Wrapper, metaclass=abc.ABCMeta):
"""
Abstract base class for Galaxy datasets.
"""
BASE_ATTRS = Wrapper.BASE_ATTRS + (
'data_type', 'file_ext', 'file_name', 'file_size', 'genome_build', 'misc_info', 'state'
)
POLLING_INTERVAL = 1 # for state monitoring
@abc.abstractmethod
def __init__(self, ds_dict, container, gi=None):
super().__init__(ds_dict, gi=gi)
object.__setattr__(self, 'container', container)
@property
def container_id(self):
"""