mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
John Chilton
b4c77b7617
This is a slight tightening up of the ad-hoc, experimental workflow YAML definition used by the test framework (and by Kyle, but lets just admit Kyle is part of Galaxy's test framework). This format is still defined entirely client side by transcoding the YAML or python object description into real (or format 1) Galaxy workflows - so these cannot realisticaly be declared part of Galaxy's public interface and can remain experimental and subject to change. In addition to refactoring the code implementing these workflows for upstream modifications to explore new features, the format itself has been made slightly more stringent in two ways: - 'steps' must now be explicit (previously converted auto-convert list to dict). - Ensure the workflow declared a 'class' is defined and the class is 'GalaxyWorkflow'. This will help with discovery of workflow objects with downstream tooling (planemo for workflows?) and brings the Galaxy definition slightly more inline with the CWL definition for workflows (very slightly).
54 lines
1.3 KiB
Python
54 lines
1.3 KiB
Python
import abc
|
|
import json
|
|
|
|
import bioblend
|
|
import six
|
|
|
|
|
|
@six.add_metaclass(abc.ABCMeta)
|
|
class ImporterGalaxyInterface(object):
|
|
""" An abstract interface describing the interaction between
|
|
Galaxy and the workflow import code.
|
|
"""
|
|
|
|
@abc.abstractmethod
|
|
def import_workflow(self, workflow):
|
|
""" Import a workflow via POST /api/workflows or
|
|
comparable interface into Galaxy.
|
|
"""
|
|
pass
|
|
|
|
|
|
class BioBlendImporterGalaxyInterface(object):
|
|
|
|
def __init__(self, **kwds):
|
|
"""
|
|
"""
|
|
url = None
|
|
|
|
user_key = None
|
|
user_gi = None
|
|
if "user_gi" in kwds:
|
|
user_gi = kwds["user_gi"]
|
|
elif "gi" in kwds:
|
|
user_gi = kwds["gi"]
|
|
elif "url" in kwds and "user_key" in kwds:
|
|
url = kwds["url"]
|
|
user_key = kwds["user_key"]
|
|
|
|
if user_gi is None:
|
|
assert url is not None
|
|
assert user_key is not None
|
|
user_gi = bioblend.GalaxyInstance(url=url, key=user_key)
|
|
|
|
self._user_gi = user_gi
|
|
|
|
def import_workflow(self, workflow):
|
|
workflow_str = json.dumps(workflow, indent=4)
|
|
return self._user_gi.workflows.import_workflow_json(
|
|
workflow_str
|
|
)
|
|
|
|
def import_tool(self, tool_representation):
|
|
pass
|