mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-24 02:04:04 +08:00
Outline how to share complex code between old and new style controllers.
More use of Pydantic models in the old code and more use of "view" functions below the controller code - mirroring inn some ways what is done for dependency management APIs (which I had structured to be very lightweight in an alternative framework such as this).
This commit is contained in:
@@ -1,7 +1,12 @@
|
||||
import json
|
||||
import logging
|
||||
import typing
|
||||
|
||||
from boltons.iterutils import remap
|
||||
from pydantic import (
|
||||
BaseModel,
|
||||
Field,
|
||||
)
|
||||
from sqlalchemy import and_, false, func, or_
|
||||
from sqlalchemy.orm import aliased
|
||||
from sqlalchemy.sql import select
|
||||
@@ -25,6 +30,10 @@ from galaxy.util import (
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class JobLock(BaseModel):
|
||||
active: bool = Field(title="Job lock status", description="If active, jobs will not dispatch")
|
||||
|
||||
|
||||
def get_path_key(path_tuple):
|
||||
path_key = ""
|
||||
tuple_elements = len(path_tuple)
|
||||
@@ -50,6 +59,13 @@ class JobManager:
|
||||
self.app = app
|
||||
self.dataset_manager = DatasetManager(app)
|
||||
|
||||
def job_lock(self):
|
||||
return JobLock(active=self.app.job_manager.job_lock)
|
||||
|
||||
def update_job_lock(self, job_lock: JobLock):
|
||||
self.app.queue_worker.send_control_task('admin_job_lock', kwargs={'job_lock': job_lock.active}, get_response=True)
|
||||
return self.job_lock()
|
||||
|
||||
def get_accessible_job(self, trans, decoded_job_id):
|
||||
job = trans.sa_session.query(trans.app.model.Job).filter(trans.app.model.Job.id == decoded_job_id).first()
|
||||
if job is None:
|
||||
@@ -368,6 +384,30 @@ class JobSearch:
|
||||
return None
|
||||
|
||||
|
||||
def view_show_job(trans, job, full: bool) -> typing.Dict:
|
||||
is_admin = trans.user_is_admin
|
||||
job_dict = trans.app.security.encode_all_ids(job.to_dict('element', system_details=is_admin), True)
|
||||
if full:
|
||||
job_dict.update(dict(
|
||||
tool_stdout=job.tool_stdout,
|
||||
tool_stderr=job.tool_stderr,
|
||||
job_stdout=job.job_stdout,
|
||||
job_stderr=job.job_stderr,
|
||||
stderr=job.stderr,
|
||||
stdout=job.stdout,
|
||||
job_messages=job.job_messages
|
||||
))
|
||||
|
||||
if is_admin:
|
||||
if job.user:
|
||||
job_dict['user_email'] = job.user.email
|
||||
else:
|
||||
job_dict['user_email'] = None
|
||||
|
||||
job_dict['job_metrics'] = summarize_job_metrics(trans, job)
|
||||
return job_dict
|
||||
|
||||
|
||||
def invocation_job_source_iter(sa_session, invocation_id):
|
||||
# TODO: Handle subworkflows.
|
||||
join = model.WorkflowInvocationStep.table.join(
|
||||
|
||||
@@ -4,6 +4,7 @@ from json import loads
|
||||
from traceback import format_exc
|
||||
|
||||
import paste.httpexceptions
|
||||
from pydantic import BaseModel
|
||||
|
||||
from galaxy.exceptions import error_codes, MessageException
|
||||
from galaxy.util import (
|
||||
@@ -326,7 +327,10 @@ def format_return_as_json(rval, jsonp_callback=None, pretty=False):
|
||||
Use `pretty=True` to return pretty printed json.
|
||||
"""
|
||||
dumps_kwargs = dict(indent=4, sort_keys=True) if pretty else {}
|
||||
json = safe_dumps(rval, **dumps_kwargs)
|
||||
if isinstance(rval, BaseModel):
|
||||
json = rval.json()
|
||||
else:
|
||||
json = safe_dumps(rval, **dumps_kwargs)
|
||||
if jsonp_callback:
|
||||
json = f"{jsonp_callback}({json});"
|
||||
return json
|
||||
|
||||
@@ -14,10 +14,6 @@ from fastapi import (
|
||||
)
|
||||
from fastapi_utils.cbv import cbv
|
||||
from fastapi_utils.inferring_router import InferringRouter as APIRouter
|
||||
from pydantic import (
|
||||
BaseModel,
|
||||
Field,
|
||||
)
|
||||
from sqlalchemy import (
|
||||
or_,
|
||||
)
|
||||
@@ -30,11 +26,13 @@ from galaxy import (
|
||||
from galaxy.app import UniverseApplication
|
||||
from galaxy.managers import hdas
|
||||
from galaxy.managers.jobs import (
|
||||
JobLock,
|
||||
JobManager,
|
||||
JobSearch,
|
||||
summarize_destination_params,
|
||||
summarize_job_metrics,
|
||||
summarize_job_parameters,
|
||||
view_show_job,
|
||||
)
|
||||
from galaxy.schema.fields import EncodedDatabaseIdField
|
||||
from galaxy.web import (
|
||||
@@ -73,11 +71,7 @@ def get_job_search(app: UniverseApplication = Depends(get_app)) -> JobSearch:
|
||||
|
||||
@lru_cache()
|
||||
def get_hda_manager(app: UniverseApplication = Depends(get_app)) -> hdas.HDAManager:
|
||||
return hdas.HDAManager(app=app)
|
||||
|
||||
|
||||
class JobLock(BaseModel):
|
||||
active: bool = Field(title="Job lock status", description="If active, jobs will not dispatch")
|
||||
return app.hda_manager
|
||||
|
||||
|
||||
@cbv(router)
|
||||
@@ -90,13 +84,12 @@ class FastAPIJobs:
|
||||
@router.get('/job_lock')
|
||||
def job_lock_status(self, admin_user=Depends(get_admin_user)) -> JobLock:
|
||||
"""Get job lock status."""
|
||||
return JobLock(active=self.trans.app.job_manager.job_lock)
|
||||
return self.job_manager.job_lock()
|
||||
|
||||
@router.put('/job_lock')
|
||||
def update_job_lock(self, admin_user=Depends(get_admin_user), job_lock: JobLock = Body(...)) -> JobLock:
|
||||
"""Set job lock status."""
|
||||
self.trans.app.queue_worker.send_control_task('admin_job_lock', kwargs={'job_lock': job_lock.active}, get_response=True)
|
||||
return JobLock(active=self.trans.app.job_manager.job_lock)
|
||||
return self.job_manager.job_lock(job_lock)
|
||||
|
||||
@router.get("/{id}")
|
||||
def show(self, id: EncodedDatabaseIdField, full: typing.Optional[bool] = False) -> typing.Dict:
|
||||
@@ -110,27 +103,7 @@ class FastAPIJobs:
|
||||
trans = self.trans
|
||||
id = trans.app.security.decode_id(id)
|
||||
job = self.job_manager.get_accessible_job(trans, id)
|
||||
is_admin = trans.user_is_admin
|
||||
job_dict = trans.app.security.encode_all_ids(job.to_dict('element', system_details=is_admin), True)
|
||||
if full:
|
||||
job_dict.update(dict(
|
||||
tool_stdout=job.tool_stdout,
|
||||
tool_stderr=job.tool_stderr,
|
||||
job_stdout=job.job_stdout,
|
||||
job_stderr=job.job_stderr,
|
||||
stderr=job.stderr,
|
||||
stdout=job.stdout,
|
||||
job_messages=job.job_messages
|
||||
))
|
||||
|
||||
if is_admin:
|
||||
if job.user:
|
||||
job_dict['user_email'] = job.user.email
|
||||
else:
|
||||
job_dict['user_email'] = None
|
||||
|
||||
job_dict['job_metrics'] = summarize_job_metrics(trans, job)
|
||||
return job_dict
|
||||
return view_show_job(trans, job, id, full)
|
||||
|
||||
|
||||
class JobController(BaseAPIController, UsesVisualizationMixin):
|
||||
@@ -241,34 +214,8 @@ class JobController(BaseAPIController, UsesVisualizationMixin):
|
||||
:returns: dictionary containing full description of job data
|
||||
"""
|
||||
job = self.__get_job(trans, id)
|
||||
is_admin = trans.user_is_admin
|
||||
job_dict = self.encode_all_ids(trans, job.to_dict('element', system_details=is_admin), True)
|
||||
full_output = util.asbool(kwd.get('full', 'false'))
|
||||
if full_output:
|
||||
if job.copied_from_job_id is not None:
|
||||
encoded_copied_from_job_id = trans.security.encode_id(job.copied_from_job_id)
|
||||
else:
|
||||
encoded_copied_from_job_id = None
|
||||
job_dict.update(dict(
|
||||
tool_stdout=job.tool_stdout,
|
||||
tool_stderr=job.tool_stderr,
|
||||
tool_version=job.tool_version,
|
||||
copied_from_job_id=encoded_copied_from_job_id,
|
||||
job_stdout=job.job_stdout,
|
||||
job_stderr=job.job_stderr,
|
||||
stderr=job.stderr,
|
||||
stdout=job.stdout,
|
||||
job_messages=job.job_messages,
|
||||
))
|
||||
|
||||
if is_admin:
|
||||
if job.user:
|
||||
job_dict['user_email'] = job.user.email
|
||||
else:
|
||||
job_dict['user_email'] = None
|
||||
|
||||
job_dict['job_metrics'] = summarize_job_metrics(trans, job)
|
||||
return job_dict
|
||||
return view_show_job(trans, job, full_output)
|
||||
|
||||
@expose_api
|
||||
def common_problems(self, trans, id, **kwd):
|
||||
@@ -588,7 +535,7 @@ class JobController(BaseAPIController, UsesVisualizationMixin):
|
||||
* GET /api/job_lock
|
||||
return boolean indicating if job lock active.
|
||||
"""
|
||||
return {"active": self.app.job_manager.job_lock}
|
||||
return self.job_manager.job_lock()
|
||||
|
||||
@require_admin
|
||||
@expose_api
|
||||
@@ -597,6 +544,6 @@ class JobController(BaseAPIController, UsesVisualizationMixin):
|
||||
* PUT /api/job_lock
|
||||
return boolean indicating if job lock active.
|
||||
"""
|
||||
job_lock = payload.get("active")
|
||||
self.app.queue_worker.send_control_task('admin_job_lock', kwargs={'job_lock': job_lock}, get_response=True)
|
||||
return {"active": self.app.job_manager.job_lock}
|
||||
active = payload.get("active")
|
||||
job_lock = JobLock(active=active)
|
||||
return self.job_manager.update_job_lock(job_lock)
|
||||
|
||||
@@ -137,6 +137,10 @@ class JobsApiTestCase(ApiTestCase, TestsTools):
|
||||
job = jobs_response.json()[0]
|
||||
job_id = job["id"]
|
||||
|
||||
job_lock_response = self._get("job_lock", admin=True)
|
||||
job_lock_response.raise_for_status()
|
||||
assert not job_lock_response.json()["active"]
|
||||
|
||||
show_jobs_response = self._get("jobs/%s" % job_id, admin=False)
|
||||
self._assert_not_has_keys(show_jobs_response.json(), "command_line", "external_id")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user