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', )