mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-19 10:51:34 +08:00
Ensure Job belongs to current SA session
This commit is contained in:
@@ -15,6 +15,8 @@ from queue import (
|
||||
Queue,
|
||||
)
|
||||
|
||||
from sqlalchemy.orm import object_session
|
||||
|
||||
import galaxy.jobs
|
||||
from galaxy import model
|
||||
from galaxy.exceptions import ConfigurationError
|
||||
@@ -154,6 +156,10 @@ class BaseJobRunner:
|
||||
name = method.__name__
|
||||
except Exception:
|
||||
name = UNKNOWN
|
||||
|
||||
# Ensure a Job object belongs to a session
|
||||
self._ensure_db_session(arg)
|
||||
|
||||
try:
|
||||
action_str = f"galaxy.jobs.runners.{self.__class__.__name__.lower()}.{name}"
|
||||
action_timer = self.app.execution_timer_factory.get_timer(
|
||||
@@ -171,6 +177,18 @@ class BaseJobRunner:
|
||||
# Prevent fail_job cycle in the work_queue
|
||||
self.work_queue.put((self.fail_job, job_state))
|
||||
|
||||
def _ensure_db_session(self, arg: typing.Union["JobWrapper", "JobState"]) -> None:
|
||||
"""Ensure Job object belongs to current session."""
|
||||
try:
|
||||
job_wrapper = arg.job_wrapper # type: ignore[union-attr]
|
||||
except AttributeError:
|
||||
job_wrapper = arg
|
||||
|
||||
if job_wrapper._job_io:
|
||||
job = job_wrapper._job_io.job
|
||||
if object_session(job) is None:
|
||||
self.app.model.session().add(job)
|
||||
|
||||
# Causes a runner's `queue_job` method to be called from a worker thread
|
||||
def put(self, job_wrapper: "MinimalJobWrapper"):
|
||||
"""Add a job to the queue (by job identifier), indicate that the job is ready to run."""
|
||||
|
||||
Reference in New Issue
Block a user