diff --git a/beagle/__init__.py b/beagle/__init__.py index 55e470907..3d67cd6bb 100644 --- a/beagle/__init__.py +++ b/beagle/__init__.py @@ -1 +1 @@ -__version__ = "2.3.0" +__version__ = "2.4.0" diff --git a/beagle/settings.py b/beagle/settings.py index 61a3be5fa..967593ae1 100644 --- a/beagle/settings.py +++ b/beagle/settings.py @@ -459,6 +459,9 @@ BEAGLE_NOTIFIER_EMAIL_ABOUT_NEW_USERS = os.environ.get("BEAGLE_NOTIFIER_EMAIL_ABOUT_NEW_USERS") BEAGLE_NOTIFIER_EMAIL_FROM = os.environ.get("BEAGLE_NOTIFIER_EMAIL_FROM") +SMTP_HOST = os.environ.get("SMTP_HOST", "localhost") +SMTP_PORT = int(os.environ.get("SMTP_PORT", "25")) + BEAGLE_NOTIFIER_VOYAGER_STATUS_EMAIL_TO = os.environ.get("BEAGLE_NOTIFIER_VOYAGER_STATUS_EMAIL_TO", "").split(",") BEAGLE_NOTIFIER_VOYAGER_STATUS_BLACKLIST = os.environ.get("BEAGLE_NOTIFIER_VOYAGER_STATUS_BLACKLIST", "").split(",") BEAGLE_NOTIFIER_VOYAGER_STATUS_PIPELINES = os.environ.get("BEAGLE_NOTIFIER_VOYAGER_STATUS_PIPELINES", "").split(",") @@ -488,6 +491,7 @@ FASTQ_DEFAULT_LOCATION_PREFIX = os.environ.get("BEAGLE_FASTQ_DEFAULT_LOCATION_PREFIX") FASTQ_IRIS_LOCATION_PREFIX = os.environ.get("BEAGLE_FASTQ_IRIS_LOCATION_PREFIX") +FASTQ_DEFAULT_STAGING_PATH = os.environ.get("BEAGLE_FASTQ_DEFAULT_STAGING_PATH", "") DEFAULT_LOG_PREFIX = os.environ.get("BEAGLE_DEFAULT_LOG_PREFIX", "") DEFAULT_LOG_PATH = os.environ.get("BEAGLE_DEFAULT_LOG_PATH", "/tmp") diff --git a/beagle_etl/admin.py b/beagle_etl/admin.py index da0ae1689..dc5a8b1c2 100644 --- a/beagle_etl/admin.py +++ b/beagle_etl/admin.py @@ -1,5 +1,6 @@ from django.contrib import admin from django.contrib.admin import ModelAdmin +from django.conf import settings from lib.admin import link_relation from .models import ( Operator, @@ -7,6 +8,7 @@ SMILEMessage, RequestCallbackJob, ) +from .jobs.metadb_jobs import new_request from advanced_filters.admin import AdminAdvancedFiltersMixin @@ -30,8 +32,16 @@ class SMILEMessagesAdmin(AdminAdvancedFiltersMixin, ModelAdmin): list_filter = ("request_id", "topic", "status") advanced_filter_fields = ("request_id", "topic", "status") ordering = ("-created_date",) + list_display = ("created_date", "request_id", "gene_panel", "topic", "status") search_fields = ("request_id", "gene_panel") + actions = ["force_import"] + + @admin.action(description="Force import selected SMILE messages (skip validation)") + def force_import(self, request, queryset): + for message in queryset.filter(topic=settings.METADB_NATS_NEW_REQUEST): + new_request.delay(str(message.id), force_import=True) + self.message_user(request, f"Force import triggered for {queryset.count()} message(s).") class RequestCallbackJobAdmin(ModelAdmin): diff --git a/beagle_etl/celery.py b/beagle_etl/celery.py index d6d6145ff..d7b58feb6 100644 --- a/beagle_etl/celery.py +++ b/beagle_etl/celery.py @@ -50,6 +50,7 @@ def setup_task_logger(logger, *args, **kwargs): "beagle_etl.tasks.job_processor": {"queue": settings.BEAGLE_DEFAULT_QUEUE}, "beagle_etl.tasks.process_smile_events": {"queue": settings.BEAGLE_DEFAULT_QUEUE}, "beagle_etl.tasks.process_job_with_lock": {"queue": settings.BEAGLE_DEFAULT_QUEUE}, + "beagle_etl.jobs.metadb_jobs.new_request": {"queue": settings.BEAGLE_DEFAULT_QUEUE}, "beagle_etl.jobs.metadb_jobs.update_job": {"queue": settings.BEAGLE_DEFAULT_QUEUE}, "beagle_etl.jobs.metadb_jobs.not_supported": {"queue": settings.BEAGLE_DEFAULT_QUEUE}, "beagle_etl.jobs.metadb_jobs.request_callback": {"queue": settings.BEAGLE_DEFAULT_QUEUE}, diff --git a/beagle_etl/jobs/metadb_jobs.py b/beagle_etl/jobs/metadb_jobs.py index 570f37dee..d125aedfd 100644 --- a/beagle_etl/jobs/metadb_jobs.py +++ b/beagle_etl/jobs/metadb_jobs.py @@ -188,17 +188,17 @@ def request_update_notification(request_id): @shared_task -def new_request(message_id): +def new_request(message_id, force_import=False): message = SMILEMessage.objects.get(id=message_id) try: - data = RequestMetadata.from_dict(json.loads(message.message)) + data = RequestMetadata.from_dict(json.loads(message.message), force_import=force_import) except Exception as e: message.add_log(str(e)) message.failed() return - if not data.isCmoRequest: + if not data.isCmoRequest and not force_import: # Non CmoRequests not supported logger.info(f"Request {data.igoRequestId} is not CMO Request") message.add_log(f"Request {data.igoRequestId} is not CMO Request") diff --git a/beagle_etl/serializers.py b/beagle_etl/serializers.py index c2325dac6..702e3f2ea 100644 --- a/beagle_etl/serializers.py +++ b/beagle_etl/serializers.py @@ -1,5 +1,5 @@ from rest_framework import serializers -from .models import ETLConfiguration +from .models import ETLConfiguration, SMILEMessage, SmileMessageStatus def ValidateDict(value): @@ -29,3 +29,21 @@ class AssayUpdateSerializer(serializers.Serializer): class RequestIdLimsPullSerializer(serializers.Serializer): request_ids = serializers.ListField(child=serializers.CharField(max_length=30)) redelivery = serializers.BooleanField(default=False) + + +class SMILEMessageSerializer(serializers.ModelSerializer): + status = serializers.SerializerMethodField() + + class Meta: + model = SMILEMessage + fields = "__all__" + + def get_status(self, obj): + return SmileMessageStatus(obj.status).name + + +class SMILEMessageListSerializer(serializers.Serializer): + request_id = serializers.CharField(required=False) + topic = serializers.CharField(required=False) + gene_panel = serializers.CharField(required=False) + status = serializers.ChoiceField([(s.name, s.value) for s in SmileMessageStatus], allow_blank=True, required=False) diff --git a/beagle_etl/smile_message/objects/request_object.py b/beagle_etl/smile_message/objects/request_object.py index 3a6e332a0..9ce7d1e9a 100644 --- a/beagle_etl/smile_message/objects/request_object.py +++ b/beagle_etl/smile_message/objects/request_object.py @@ -41,15 +41,15 @@ class RequestMetadata: pooledNormals: Optional[List[str]] = None @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "RequestMetadata": + def from_dict(cls, data: Dict[str, Any], force_import: bool = False) -> "RequestMetadata": """Deserialize from dictionary.""" # Handle nested status status_data = data.get("status", {}) status = RequestStatus(**status_data) if status_data else RequestStatus(False, "{}") # Handle nested samples - samples_data = data.get("samples", []) - samples = [SampleMetadata.from_dict(sample) for sample in samples_data] + samples_data = data.get("samples") or [] + samples = [SampleMetadata.from_dict(sample, force_import=force_import) for sample in samples_data] # Handle delivery date conversion delivery_date = None diff --git a/beagle_etl/smile_message/objects/sample_object.py b/beagle_etl/smile_message/objects/sample_object.py index 6b1dbc5e3..47e373649 100644 --- a/beagle_etl/smile_message/objects/sample_object.py +++ b/beagle_etl/smile_message/objects/sample_object.py @@ -1,6 +1,6 @@ import re import logging -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import List, Optional, Dict, Any from django.conf import settings from beagle_etl.exceptions import ( @@ -217,9 +217,10 @@ class SampleMetadata: tubeId: Optional[str] = None cfDNA2dBarcode: Optional[str] = None cmoInfoIgoId: Optional[str] = None + skip_validation: bool = field(default=False, compare=False, repr=False) @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "SampleMetadata": + def from_dict(cls, data: Dict[str, Any], force_import: bool = False) -> "SampleMetadata": """Deserialize SampleMetadata from dictionary.""" # Handle nested status status_data = data.get("status", {}) @@ -227,18 +228,23 @@ def from_dict(cls, data: Dict[str, Any]) -> "SampleMetadata": # Handle nested cmoSampleIdFields cmo_fields_data = data.get("cmoSampleIdFields", {}) - cmo_fields = CmoSampleIdFields(**cmo_fields_data) if cmo_fields_data else CmoSampleIdFields("", "", "", "") + cmo_fields = CmoSampleIdFields( + naToExtract=cmo_fields_data.get("naToExtract", ""), + normalizedPatientId=cmo_fields_data.get("normalizedPatientId", ""), + sampleType=cmo_fields_data.get("sampleType", ""), + recipe=cmo_fields_data.get("recipe", ""), + ) # Handle libraries - libraries_data = data.get("libraries", []) + libraries_data = data.get("libraries") or [] libraries = [Library.from_dict(lib) for lib in libraries_data] # Handle sample aliases - sample_aliases_data = data.get("sampleAliases", []) + sample_aliases_data = data.get("sampleAliases") or [] sample_aliases = [SampleAlias(**alias) for alias in sample_aliases_data] # Handle patient aliases - patient_aliases_data = data.get("patientAliases", []) + patient_aliases_data = data.get("patientAliases") or [] patient_aliases = [PatientAlias(**alias) for alias in patient_aliases_data] return cls( @@ -270,14 +276,17 @@ def from_dict(cls, data: Dict[str, Any]) -> "SampleMetadata": igoComplete=data.get("igoComplete"), status=status, cmoSampleIdFields=cmo_fields, - qcReports=data.get("qcReports", []), + qcReports=data.get("qcReports") or [], libraries=libraries, sampleAliases=sample_aliases, patientAliases=patient_aliases, additionalProperties=data.get("additionalProperties", {}), + skip_validation=force_import, ) def __post_init__(self): + if self.skip_validation: + return self._validate_primary_id() required = PANEL_REQUIRED_FIELDS.get(self.genePanel, DEFAULT_REQUIRED_FIELDS) self._validate_required_fields(required) diff --git a/beagle_etl/urls.py b/beagle_etl/urls.py index ca232e417..ee55eca07 100644 --- a/beagle_etl/urls.py +++ b/beagle_etl/urls.py @@ -1,6 +1,6 @@ from rest_framework import routers from django.urls import path, include -from beagle_etl.views import AssayViewSet +from beagle_etl.views import AssayViewSet, ForceImportView, SMILEMessageViewSet router = routers.DefaultRouter() @@ -8,4 +8,6 @@ urlpatterns = [ path("", include(router.urls)), path("assay", AssayViewSet.as_view()), + path("import/messages/", SMILEMessageViewSet.as_view({"get": "list"})), + path("import//", ForceImportView.as_view()), ] diff --git a/beagle_etl/views.py b/beagle_etl/views.py index b16b6c718..865a2b1c9 100644 --- a/beagle_etl/views.py +++ b/beagle_etl/views.py @@ -1,15 +1,68 @@ -from rest_framework import status +from django.conf import settings +from rest_framework import mixins, status from rest_framework.response import Response from rest_framework.generics import GenericAPIView -from beagle_etl.models import ETLConfiguration +from rest_framework.views import APIView +from rest_framework.viewsets import GenericViewSet +from rest_framework.permissions import IsAuthenticated +from beagle_etl.models import ETLConfiguration, SMILEMessage, SmileMessageStatus +from beagle_etl.jobs.metadb_jobs import new_request from drf_yasg.utils import swagger_auto_schema from .serializers import ( AssaySerializer, AssayElementSerializer, AssayUpdateSerializer, + SMILEMessageSerializer, + SMILEMessageListSerializer, ) +class ForceImportView(APIView): + permission_classes = (IsAuthenticated,) + + def post(self, _request, request_id): + message = ( + SMILEMessage.objects.filter(request_id=request_id, topic=settings.METADB_NATS_NEW_REQUEST) + .order_by("-created_date") + .first() + ) + if not message: + return Response( + {"detail": f"No new-request SMILEMessage found for request_id {request_id}."}, + status=status.HTTP_404_NOT_FOUND, + ) + new_request.delay(str(message.id), force_import=True) + return Response( + {"detail": f"Force import triggered for request {request_id} (message {message.id})."}, + status=status.HTTP_202_ACCEPTED, + ) + + +class SMILEMessageViewSet(mixins.ListModelMixin, GenericViewSet): + queryset = SMILEMessage.objects.order_by("-created_date").all() + serializer_class = SMILEMessageListSerializer + permission_classes = (IsAuthenticated,) + + @swagger_auto_schema(query_serializer=SMILEMessageListSerializer) + def list(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.query_params) + if not serializer.is_valid(): + return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) + validated = serializer.validated_data + queryset = self.queryset + if validated.get("request_id"): + queryset = queryset.filter(request_id=validated["request_id"]) + if validated.get("topic"): + queryset = queryset.filter(topic=validated["topic"]) + if validated.get("gene_panel"): + queryset = queryset.filter(gene_panel=validated["gene_panel"]) + if validated.get("status"): + queryset = queryset.filter(status=SmileMessageStatus[validated["status"]].value) + page = self.paginate_queryset(queryset) + serializer = SMILEMessageSerializer(page, many=True) + return self.get_paginated_response(serializer.data) + + class AssayViewSet(GenericAPIView): serializer_class = AssaySerializer queryset = ETLConfiguration.objects.all() diff --git a/compose.yaml b/compose.yaml index 92d7abc17..344f31472 100644 --- a/compose.yaml +++ b/compose.yaml @@ -592,6 +592,14 @@ services: beagle_pgbouncer: condition: service_healthy restart: false + postfix: + image: boky/postfix:5.1.0-debian + container_name: ${BEAGLE_DEPLOYMENT}-postfix + restart: always + networks: + - voyager_net + environment: + - ALLOWED_SENDER_DOMAINS=mskcc.org volumes: postgres_path: driver: local diff --git a/file_manager/copy_service/copy_service.py b/file_manager/copy_service/copy_service.py index ef28fe32b..204c7f127 100644 --- a/file_manager/copy_service/copy_service.py +++ b/file_manager/copy_service/copy_service.py @@ -25,19 +25,25 @@ def copy(path_from, path_to): os.chmod(path_to, settings.COPY_FILE_PERMISSION) @staticmethod - def remap(gene_panel, path, mapping=settings.DEFAULT_MAPPING): - prefix, dst = CopyService._get_mapping(gene_panel, path, mapping) + def remap(gene_panel, path, file_group=settings.IMPORT_FILE_GROUP, mapping=settings.DEFAULT_MAPPING): + prefix, dst = CopyService._get_mapping(gene_panel, path, file_group, mapping) if prefix and dst: path = path.replace(prefix, dst) logger.info("New path {path}".format(path=path)) return path @staticmethod - def _get_mapping(gene_panel, path, mapping=settings.DEFAULT_MAPPING): - recipe_mapping = mapping.get(gene_panel, {}) - for prefix, dst in recipe_mapping.items(): - if path.startswith(prefix): - return prefix, dst + def _get_mapping(gene_panel, path, file_group=settings.IMPORT_FILE_GROUP, mapping=settings.DEFAULT_MAPPING): + if file_group == settings.IMPORT_FILE_GROUP: + recipe_mapping = mapping.get(gene_panel, {}) + for prefix, dst in recipe_mapping.items(): + if path.startswith(prefix): + return prefix, dst + else: + return ( + settings.FASTQ_IRIS_LOCATION_PREFIX, + os.path.join(settings.FASTQ_DEFAULT_STAGING_PATH, file_group) + "/", + ) return None, None @staticmethod diff --git a/file_manager/file_manager/file_manager.py b/file_manager/file_manager/file_manager.py index a9a15abfd..3020b5446 100644 --- a/file_manager/file_manager/file_manager.py +++ b/file_manager/file_manager/file_manager.py @@ -29,7 +29,7 @@ def stage_sample(self, sample_id): files_to_stage = 0 for f in files: if not f.file.is_available: - new_path = CopyService.remap(gene_panel, f.file.path) + new_path = CopyService.remap(gene_panel, f.file.path, str(f.file.file_group.id)) if new_path != f.file.path: files_to_stage += 1 @@ -55,7 +55,7 @@ def stage_file(self, file_obj, gene_panel, sample_job=None): Returns: Task signature or None """ if not file_obj.is_available: - new_path = CopyService.remap(gene_panel, file_obj.path) + new_path = CopyService.remap(gene_panel, file_obj.path, str(file_obj.file_group.id)) if new_path != file_obj.path: fp_job, created = FileProviderJob.objects.provide_file( file_obj, file_obj.path, new_path, sample_job=sample_job diff --git a/file_manager/tests/copy_service/test_copy_service.py b/file_manager/tests/copy_service/test_copy_service.py index 2318f946d..14ec08cdc 100644 --- a/file_manager/tests/copy_service/test_copy_service.py +++ b/file_manager/tests/copy_service/test_copy_service.py @@ -1,4 +1,7 @@ -from django.test import TestCase +import os + +from django.test import TestCase, override_settings + from file_manager.copy_service.copy_service import CopyService @@ -11,12 +14,12 @@ def setUp(self): def test_remap(self): old_path = "/path/to/file/file1.fastq" - new_path = CopyService.remap(self.recipe, old_path, self.mapping) + new_path = CopyService.remap(self.recipe, old_path, mapping=self.mapping) self.assertEqual(new_path, "/new/path/to/file/file1.fastq") def test_remap_no_mapping(self): old_path = "/some/other/path/to/file/file1.fastq" - new_path = CopyService.remap(self.recipe, old_path, self.mapping) + new_path = CopyService.remap(self.recipe, old_path, mapping=self.mapping) self.assertEqual(new_path, old_path) def test_remap_multiple_prefixes(self): @@ -29,11 +32,11 @@ def test_remap_multiple_prefixes(self): } path1 = "/path/to/source1/file1.fastq" - new_path1 = CopyService.remap(self.recipe, path1, mapping) + new_path1 = CopyService.remap(self.recipe, path1, mapping=mapping) self.assertEqual(new_path1, "/staging/dest1/file1.fastq") path2 = "/path/to/source2/file2.fastq" - new_path2 = CopyService.remap(self.recipe, path2, mapping) + new_path2 = CopyService.remap(self.recipe, path2, mapping=mapping) self.assertEqual(new_path2, "/staging/dest2/file2.fastq") def test_remap_different_recipe(self): @@ -44,24 +47,45 @@ def test_remap_different_recipe(self): } path = "/path/to/file.fastq" - new_path_impact = CopyService.remap("IMPACT468", path, mapping) + new_path_impact = CopyService.remap("IMPACT468", path, mapping=mapping) self.assertEqual(new_path_impact, "/staging/impact/file.fastq") - new_path_heme = CopyService.remap("HEMEPACT", path, mapping) + new_path_heme = CopyService.remap("HEMEPACT", path, mapping=mapping) self.assertEqual(new_path_heme, "/staging/heme/file.fastq") def test_get_mapping(self): """Test internal _get_mapping method""" - prefix, dst = CopyService._get_mapping(self.recipe, "/path/to/file.fastq", self.mapping) + prefix, dst = CopyService._get_mapping(self.recipe, "/path/to/file.fastq", mapping=self.mapping) self.assertEqual(prefix, "/path/to") self.assertEqual(dst, "/new/path/to") def test_get_mapping_no_match(self): """Test _get_mapping when no prefix matches""" - prefix, dst = CopyService._get_mapping(self.recipe, "/other/path/file.fastq", self.mapping) + prefix, dst = CopyService._get_mapping(self.recipe, "/other/path/file.fastq", mapping=self.mapping) self.assertIsNone(prefix) self.assertIsNone(dst) + @override_settings(FASTQ_IRIS_LOCATION_PREFIX="/igo/delivery", FASTQ_DEFAULT_STAGING_PATH="/staging") + def test_get_mapping_other_file_group(self): + """Test _get_mapping when file_group is not IMPORT_FILE_GROUP: it should stage under + FASTQ_DEFAULT_STAGING_PATH/, regardless of the recipe mapping.""" + other_file_group = "some-other-file-group-id" + prefix, dst = CopyService._get_mapping( + self.recipe, "/test/delivery/file.fastq", file_group=other_file_group, mapping=self.mapping + ) + self.assertEqual(prefix, "/igo/delivery") + self.assertEqual(dst, os.path.join("/staging", other_file_group) + "/") + + @override_settings(FASTQ_IRIS_LOCATION_PREFIX="/igo/delivery", FASTQ_DEFAULT_STAGING_PATH="/staging") + def test_remap_other_file_group(self): + """Test that remap stages a file under FASTQ_DEFAULT_STAGING_PATH/ + when file_group is different from IMPORT_FILE_GROUP.""" + other_file_group = "some-other-file-group-id" + old_path = "/test/delivery/file/file1.fastq" + new_path = CopyService.remap(self.recipe, old_path, file_group=other_file_group, mapping=self.mapping) + expected_dst = os.path.join("/staging", other_file_group + "/") + self.assertEqual(new_path, old_path.replace("/igo/delivery", expected_dst)) + def test_get_reverse_mapping(self): """Test reverse mapping to convert staged path back to original""" staged_path = "/new/path/to/file/file1.fastq" @@ -81,7 +105,7 @@ def test_reverse_mapping_symmetry(self): original_path = "/path/to/subdir/file.fastq" # Forward mapping - staged_path = CopyService.remap(self.recipe, original_path, self.mapping) + staged_path = CopyService.remap(self.recipe, original_path, mapping=self.mapping) self.assertEqual(staged_path, "/new/path/to/subdir/file.fastq") # Reverse mapping should give us back the components diff --git a/notifier/email/email_client.py b/notifier/email/email_client.py index 167f8d4fc..dd833a213 100644 --- a/notifier/email/email_client.py +++ b/notifier/email/email_client.py @@ -5,6 +5,8 @@ from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart +from django.conf import settings + class EmailClient(object): logger = logging.getLogger(__name__) @@ -15,12 +17,13 @@ def __init__(self, email_to, email_from, subject, content): self.content = content self.email_from = email_from self.domain = "mskcc.org" - self.SMTP_server = "localhost" + self.SMTP_server = settings.SMTP_HOST + self.SMTP_port = settings.SMTP_PORT def send(self): server = None try: - server = smtplib.SMTP(self.SMTP_server) + server = smtplib.SMTP(self.SMTP_server, self.SMTP_port) msg = MIMEMultipart("alternative") msg["Subject"] = self.subject msg["From"] = self.email_from diff --git a/requirements.txt b/requirements.txt index 79a1017c8..c7cecc2c8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -Django==6.0.4 +Django==6.0.7 psycopg[binary]==3.2.3 djangorestframework==3.17.1 markdown==3.1.1 @@ -9,7 +9,7 @@ django-cors-headers==4.9.0 django-celery-results==2.6.0 python-slugify==3.0.2 drf-yasg==1.21.8 -GitPython==3.1.37 +GitPython==3.1.58 jsonschema==4.23.0 setuptools==75.8.0 deepdiff==4.0.7 diff --git a/runner/tasks.py b/runner/tasks.py index 207ddaeb7..ebdf1ac2e 100644 --- a/runner/tasks.py +++ b/runner/tasks.py @@ -56,7 +56,7 @@ def stage_files_for_operator( ): staging_tasks = [] try: - staging_tasks, sample_jobs = stage_files(request_id, pairing, job_group_id) + staging_tasks, sample_jobs = stage_files(request_id, pairing, job_group_id, file_group_id) except Exception as e: logger.warning(format_log(f"Failed to stage files: {str(e)}", job_group_id=job_group_id)) @@ -251,7 +251,7 @@ def create_operator_run_from_jobs( operator_run.save() -def stage_files(request_id=None, pairing=None, job_group_id=None): +def stage_files(request_id=None, pairing=None, job_group_id=None, file_group_id=None): """ Stage files and return list of staging task signatures. Returns (staging_tasks, sample_jobs) where: @@ -278,7 +278,7 @@ def stage_files(request_id=None, pairing=None, job_group_id=None): logger.info(format_log("No samples to stage", job_group_id=job_group_id)) return staging_tasks, sample_jobs - file_manager = FileManager() + file_manager = FileManager(file_group_id) if file_group_id else FileManager() for sample in samples: logger.info(format_log(f"Staging files for sample {sample}", request_id=request_id, job_group_id=job_group_id)) sample_job, task_sigs = file_manager.stage_sample(sample)