Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.jobs.runners.util.drmaa

try:
    from drmaa import JobControlAction, Session
except ImportError as e:
    # Will not be able to use DRMAA
    Session = None

NO_DRMAA_MESSAGE = "Attempt to use DRMAA, but DRMAA Python library cannot be loaded."


[docs]class DrmaaSessionFactory(object): """ Abstraction used to production DrmaaSession wrappers. """
[docs] def __init__(self): self.session_constructor = Session
[docs] def get(self, **kwds): session_constructor = self.session_constructor if not session_constructor: raise Exception(NO_DRMAA_MESSAGE) return DrmaaSession(session_constructor(), **kwds)
class DrmaaSession(object): """ Abstraction around `drmaa` module `Session` objects. """ def __init__(self, session, **kwds): self.session = session session.initialize() def run_job(self, **kwds): """ Create a DRMAA job template, populate with specified properties, run the job, and return the external_job_id. """ template = self.session.createJobTemplate() try: for key in kwds: setattr(template, key, kwds[key]) return self.session.runJob(template) finally: self.session.deleteJobTemplate(template) def kill(self, external_job_id): return self.session.control(str(external_job_id), JobControlAction.TERMINATE) def job_status(self, external_job_id): return self.session.jobStatus(str(external_job_id)) def close(self): return self.session.exit() __all__ = ('DrmaaSessionFactory', )