Merge pull request #417 from AllenNeuralDynamics/release-v2.1.1

Release v2.1.1

Author: mekhlakapoor

src/aind_data_transfer_service/__init__.py

@@ -8,7 +8,7 @@
 import yaml
 from pythonjsonlogger import json as log_json
 
-__version__ = "2.1.0"
+__version__ = "2.1.1"
 
 
 # We want to standardize the timestamp format to UTC and ISO-8601, which

src/aind_data_transfer_service/models/core.py

@@ -141,7 +141,8 @@ def __init__(self, /, **data: Any) -> None:
     user_email: Optional[EmailStr] = Field(
         default=None,
         description=(
-            "Optional email address to receive job status notifications"
+            "Email address to receive job status notifications. "
+            "If not set, will use email from SubmitJobRequestV2."
         ),
     )
     email_notification_types: Optional[
@@ -284,10 +285,10 @@ def __init__(self, /, **data: Any) -> None:
     model_config = ConfigDict(use_enum_values=True, extra="ignore")
 
     dag_id: Literal["transform_and_upload_v2"] = "transform_and_upload_v2"
-    user_email: EmailStr = Field(
-        ...,
+    user_email: Optional[EmailStr] = Field(
+        default=None,
         description=(
-            "Required email address to receive job status notifications"
+            "Email address to use if not set in individual job configs"
         ),
     )
     email_notification_types: Set[
@@ -311,7 +312,13 @@ def propagate_email_settings(self):
         global_email_user = self.user_email
         global_email_notification_types = self.email_notification_types
         for upload_job in self.upload_jobs:
-            if global_email_user is not None and upload_job.user_email is None:
+            if upload_job.user_email is None and global_email_user is None:
+                raise ValueError(
+                    f"No user_email set for job {upload_job.s3_prefix}. "
+                    "Either set user_email in the job config or in the "
+                    "SubmitJobRequestV2."
+                )
+            elif upload_job.user_email is None:
                 upload_job.user_email = global_email_user
             if upload_job.email_notification_types is None:
                 upload_job.email_notification_types = (

tests/resources/legacy_sample.csv

@@ -1,4 +1,4 @@
-project_name, modality0, modality0.source, modality1, modality1.source, s3-bucket, subject-id, platform, acq-datetime, force_cloud_sync, modality0.process_capsule_id, job_type
-Ephys Platform, ecephys, dir/data_set_1, ,, private, 123454, ecephys, 2020-10-10 14:10:10, True,, default
-Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open, 123456, behavior, 10/13/2020 1:10:10 PM,,1f999652-00a0-4c4b-99b5-64c2985ad070, default
-Behavior Platform, behavior-videos, dir/data_set_2, behavior-videos, dir/data_set_3, default, 123456, behavior, 10/13/2020 1:10:10 PM,,, default
+project_name, modality0, modality0.source, modality1, modality1.source, s3-bucket, subject-id, platform, acq-datetime, force_cloud_sync, modality0.process_capsule_id, job_type, user_email
+Ephys Platform, ecephys, dir/data_set_1, ,, private, 123454, ecephys, 2020-10-10 14:10:10, True,, default, test@example.com
+Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open, 123456, behavior, 10/13/2020 1:10:10 PM,,1f999652-00a0-4c4b-99b5-64c2985ad070, default, test@example.com
+Behavior Platform, behavior-videos, dir/data_set_2, behavior-videos, dir/data_set_3, default, 123456, behavior, 10/13/2020 1:10:10 PM,,, default, test@example.com

tests/resources/legacy_sample2.csv

@@ -1,2 +1,2 @@
-project_name, modality0, modality0.source, modality1, modality1.source, s3-bucket, subject-id, platform, acq-datetime, modality0.pipeline_id, job_type
-Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open, 123456, behavior, 10/13/2020 1:10:10 PM,1f999652-00a0-4c4b-99b5-64c2985ad070, default
+project_name, modality0, modality0.source, modality1, modality1.source, s3-bucket, subject-id, platform, acq-datetime, modality0.pipeline_id, job_type, user_email
+Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open, 123456, behavior, 10/13/2020 1:10:10 PM,1f999652-00a0-4c4b-99b5-64c2985ad070, default, test@example.com

tests/resources/nested_sample.csv

@@ -1,2 +1,2 @@
-project_name, modality0, modality0.input_source, modality0.compression_requested.compression_enum, subject-id, platform, acq-datetime, job_type
-Ephys Platform, ecephys, dir/data_set_1, gamma fix colorspace, 123454, ecephys, 2020-10-10 14:10:10, ecephys
+project_name, modality0, modality0.input_source, modality0.compression_requested.compression_enum, subject-id, platform, acq-datetime, job_type, user_email
+Ephys Platform, ecephys, dir/data_set_1, gamma fix colorspace, 123454, ecephys, 2020-10-10 14:10:10, ecephys, test@example.com

tests/resources/new_sample.csv

@@ -1,4 +1,4 @@
-project_name, modality0, modality0.input_source, modality1, modality1.input_source, s3-bucket, subject-id, platform, acq-datetime, job_type, metadata_dir
-Ephys Platform, ecephys, dir/data_set_1, , , default,123454, ecephys, 2020-10-10 14:10:10, ecephys, dir/metadata
-Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open,123456, behavior, 2020-10-13T13:10:10, custom,
-Behavior Platform, behavior-videos, dir/data_set_2, , , ,123456, behavior, 2020-10-13 13:10:10, default,
+project_name, modality0, modality0.input_source, modality1, modality1.input_source, s3-bucket, subject-id, platform, acq-datetime, job_type, metadata_dir, user_email
+Ephys Platform, ecephys, dir/data_set_1, , , default,123454, ecephys, 2020-10-10 14:10:10, ecephys, dir/metadata, test@example.com
+Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open,123456, behavior, 2020-10-13T13:10:10, custom,, test@example.com
+Behavior Platform, behavior-videos, dir/data_set_2, , , ,123456, behavior, 2020-10-13 13:10:10, default,, test@example.com

tests/resources/sample_empty_rows_2.csv

@@ -1,7 +1,7 @@
-project_name, modality0, modality0.input_source, modality1, modality1.input_source, s3-bucket, subject-id, platform, acq-datetime, job_type, metadata_dir
-Ephys Platform, ecephys, dir/data_set_1, , , default,123454, ecephys, 2020-10-10 14:10:10, ecephys, dir/metadata
-Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open,123456, behavior, 2020-10-13T13:10:10, custom,
-Behavior Platform, behavior-videos, dir/data_set_2, , , ,123456, behavior, 2020-10-13 13:10:10, default,
-,,,,,,,,,,
-,,,,,,,,,,
-,,,,,,,,,,
+project_name, modality0, modality0.input_source, modality1, modality1.input_source, s3-bucket, subject-id, platform, acq-datetime, job_type, metadata_dir, user_email
+Ephys Platform, ecephys, dir/data_set_1, , , default,123454, ecephys, 2020-10-10 14:10:10, ecephys, dir/metadata, test@example.com
+Behavior Platform, behavior-videos, dir/data_set_2, MRI, dir/data_set_3, open,123456, behavior, 2020-10-13T13:10:10, custom,, test@example.com
+Behavior Platform, behavior-videos, dir/data_set_2, , , ,123456, behavior, 2020-10-13 13:10:10, default,, test@example.com
+,,,,,,,,,,,
+,,,,,,,,,,,
+,,,,,,,,,,,

tests/test_core.py

@@ -89,6 +89,7 @@ def setUpClass(cls) -> None:
         """Set up test class"""
         example_configs = UploadJobConfigsV2(
             job_type="default",
+            user_email="test@example.com",
             project_name="Behavior Platform",
             platform=Platform.BEHAVIOR,
             modalities=[Modality.BEHAVIOR_VIDEOS],
@@ -315,6 +316,7 @@ def setUpClass(cls) -> None:
         """Set up example configs to be used in tests"""
         example_upload_config = UploadJobConfigsV2(
             job_type="default",
+            user_email="test@example.com",
             project_name="Behavior Platform",
             platform=Platform.BEHAVIOR,
             modalities=[
@@ -351,7 +353,7 @@ def test_max_length(self):
         with self.assertRaises(ValidationError) as e:
             SubmitJobRequestV2(
                 upload_jobs=[upload_job for _ in range(0, 51)],
-                user_email="abc@example.com"
+                user_email="abc@example.com",
             )
         expected_message = (
             "List should have at most 50 items after validation, not 51"
@@ -366,8 +368,7 @@ def test_default_settings(self):
             **self.example_upload_config.model_dump(round_trip=True)
         )
         job_settings = SubmitJobRequestV2(
-            upload_jobs=[upload_job],
-            user_email="abc@example.com"
+            upload_jobs=[upload_job], user_email="abc@example.com"
         )
         self.assertEqual("abc@example.com", job_settings.user_email)
         self.assertEqual({"fail"}, job_settings.email_notification_types)
@@ -417,28 +418,29 @@ def test_email_validation(self):
         self.assertIn(expected_error_message, actual_error_message)
 
     def test_propagate_email_settings(self):
-        """Tests global email settings is propagated to individual jobs."""
+        """Tests that global email is propagated to jobs without email,
+        and email notification types are propagated to jobs without them."""
         example_job_configs = self.example_upload_config.model_dump(
             exclude={"user_email", "email_notification_types"}, round_trip=True
         )
         new_job = UploadJobConfigsV2(
-            user_email="xyz@example.org",
             email_notification_types=["all"],
             **example_job_configs,
         )
         job_settings = SubmitJobRequestV2(
-            user_email="abc@example.org",
+            user_email="global@example.com",
             email_notification_types={"begin", "fail"},
             upload_jobs=[
                 new_job,
                 UploadJobConfigsV2(**example_job_configs),
             ],
         )
+        self.assertEqual("global@example.com", job_settings.user_email)
         self.assertEqual(
-            "xyz@example.org", job_settings.upload_jobs[0].user_email
+            "global@example.com", job_settings.upload_jobs[0].user_email
         )
         self.assertEqual(
-            "abc@example.org", job_settings.upload_jobs[1].user_email
+            "global@example.com", job_settings.upload_jobs[1].user_email
         )
         self.assertEqual(
             {"all"}, job_settings.upload_jobs[0].email_notification_types
@@ -448,6 +450,44 @@ def test_propagate_email_settings(self):
             job_settings.upload_jobs[1].email_notification_types,
         )
 
+    def test_propagate_email_from_job(self):
+        """Tests that global email is set from first job if not provided."""
+        example_job_configs = self.example_upload_config.model_dump(
+            exclude={"email_notification_types"}, round_trip=True
+        )
+        job_settings = SubmitJobRequestV2(
+            user_email=None,
+            email_notification_types={"begin", "fail"},
+            upload_jobs=[
+                UploadJobConfigsV2(**example_job_configs),
+            ],
+        )
+        self.assertEqual(
+            "test@example.com", job_settings.upload_jobs[0].user_email
+        )
+
+    def test_missing_email_validation(self):
+        """Tests that validation error is raised when no email is provided."""
+        example_job_configs = self.example_upload_config.model_dump(
+            exclude={"user_email", "email_notification_types"}, round_trip=True
+        )
+        with self.assertRaises(ValidationError) as e:
+            SubmitJobRequestV2(
+                user_email=None,
+                upload_jobs=[UploadJobConfigsV2(**example_job_configs)],
+            )
+        errors = json.loads(e.exception.json())
+        self.assertEqual(1, len(errors))
+        self.assertIn(
+            "No user_email set for job",
+            errors[0]["msg"],
+        )
+        self.assertIn(
+            "Either set user_email in the job config or in the "
+            "SubmitJobRequestV2",
+            errors[0]["msg"],
+        )
+
     def test_check_duplicate_upload_jobs(self):
         """Tests that duplicate upload jobs are not allowed."""
         job_configs = self.example_upload_config.model_dump(
@@ -459,8 +499,7 @@ def test_check_duplicate_upload_jobs(self):
         ]
         with self.assertRaises(ValidationError) as e:
             SubmitJobRequestV2(
-                upload_jobs=upload_jobs,
-                user_email="abc@example.com"
+                upload_jobs=upload_jobs, user_email="abc@example.com"
             )
         errors = json.loads(e.exception.json())
         self.assertEqual(1, len(errors))
@@ -489,8 +528,7 @@ def test_check_duplicate_upload_jobs_same_prefix(self):
             }
             upload_jobs.append(UploadJobConfigsV2(**job_configs, tasks=tasks))
         submit_job_request = SubmitJobRequestV2(
-            upload_jobs=upload_jobs,
-            user_email="abc@example.com"
+            upload_jobs=upload_jobs, user_email="abc@example.com"
         )
         self.assertEqual(10, len(submit_job_request.upload_jobs))
 
@@ -504,7 +542,7 @@ def test_current_jobs_validation(self):
                 UploadJobConfigsV2(**job_configs, subject_id=subject_id)
                 for subject_id in ["123456", "123457", "123458"]
             ],
-            user_email="abc@example.com"
+            user_email="abc@example.com",
         )
         current_jobs = [
             j.model_dump(mode="json", exclude_none=True)
@@ -513,8 +551,7 @@ def test_current_jobs_validation(self):
         new_job = UploadJobConfigsV2(**job_configs, subject_id="123459")
         with validation_context({"current_jobs": current_jobs}):
             submit_job_request = SubmitJobRequestV2(
-                upload_jobs=[new_job],
-                user_email="abc@example.com"
+                upload_jobs=[new_job], user_email="abc@example.com"
             )
         self.assertEqual(1, len(submit_job_request.upload_jobs))
         self.assertEqual(
@@ -526,7 +563,7 @@ def test_current_jobs_validation_fail(self):
         """Tests job validation when an upload_job is already running."""
         submitted_job_request = SubmitJobRequestV2(
             upload_jobs=[self.example_upload_config],
-            user_email="abc@example.com"
+            user_email="abc@example.com",
         )
         current_jobs_1 = [
             j.model_dump(mode="json", exclude_none=True)
@@ -540,7 +577,7 @@ def test_current_jobs_validation_fail(self):
                 with validation_context({"current_jobs": current_jobs}):
                     SubmitJobRequestV2(
                         upload_jobs=[self.example_upload_config],
-                        user_email="abc@example.com"
+                        user_email="abc@example.com",
                     )
             err_msg = json.loads(err.exception.json())[0]["msg"]
             self.assertEqual(

tests/test_csv_handler.py

@@ -58,6 +58,7 @@ def test_map_csv_row_to_job(self):
         expected_jobs = [
             UploadJobConfigsV2(
                 project_name="Ephys Platform",
+                user_email="test@example.com",
                 s3_bucket="default",
                 platform=Platform.ECEPHYS,
                 modalities=[Modality.ECEPHYS],
@@ -77,6 +78,7 @@ def test_map_csv_row_to_job(self):
             ),
             UploadJobConfigsV2(
                 project_name="Behavior Platform",
+                user_email="test@example.com",
                 s3_bucket="open",
                 platform=Platform.BEHAVIOR,
                 modalities=[Modality.BEHAVIOR_VIDEOS, Modality.MRI],
@@ -96,6 +98,7 @@ def test_map_csv_row_to_job(self):
             ),
             UploadJobConfigsV2(
                 project_name="Behavior Platform",
+                user_email="test@example.com",
                 platform=Platform.BEHAVIOR,
                 modalities=[Modality.BEHAVIOR_VIDEOS],
                 tasks={
@@ -123,6 +126,7 @@ def test_map_nested_csv_row_to_job(self):
         expected_jobs = [
             UploadJobConfigsV2(
                 job_type="ecephys",
+                user_email="test@example.com",
                 s3_bucket="default",
                 project_name="Ephys Platform",
                 platform=Platform.ECEPHYS,
@@ -159,6 +163,7 @@ def test_map_legacy_csv_row_to_job(self):
         expected_jobs = [
             UploadJobConfigsV2(
                 job_type="default",
+                user_email="test@example.com",
                 s3_bucket="private",
                 project_name="Ephys Platform",
                 platform=Platform.ECEPHYS,
@@ -180,6 +185,7 @@ def test_map_legacy_csv_row_to_job(self):
             ),
             UploadJobConfigsV2(
                 job_type="default",
+                user_email="test@example.com",
                 s3_bucket="open",
                 project_name="Behavior Platform",
                 platform=Platform.BEHAVIOR,
@@ -217,6 +223,7 @@ def test_map_legacy_csv_row_to_job(self):
             ),
             UploadJobConfigsV2(
                 job_type="default",
+                user_email="test@example.com",
                 s3_bucket="default",
                 project_name="Behavior Platform",
                 platform=Platform.BEHAVIOR,
@@ -248,6 +255,7 @@ def test_map_old_csv_row_to_job(self):
         expected_jobs = [
             UploadJobConfigsV2(
                 job_type="default",
+                user_email="test@example.com",
                 s3_bucket="open",
                 project_name="Behavior Platform",
                 platform=Platform.BEHAVIOR,

tests/test_server.py

@@ -134,6 +134,7 @@ def setUpClass(cls) -> None:
         )
         example_configs_v2 = UploadJobConfigsV2(
             job_type=job_type,
+            user_email="test@example.com",
             project_name=project_name,
             platform=platform,
             subject_id=subject_id,
@@ -1598,8 +1599,7 @@ def test_submit_v1_v2_jobs_200(
         )
         mock_post.return_value = mock_response
         job_request_v2 = SubmitJobRequestV2(
-            upload_jobs=[self.example_configs_v2],
-            user_email="abc@example.com"
+            upload_jobs=[self.example_configs_v2], user_email="abc@example.com"
         )
         request_json_v2 = job_request_v2.model_dump(mode="json")
         with self.assertLogs(level="INFO") as captured:
@@ -1686,6 +1686,7 @@ def test_submit_v1_v2_jobs_exception_500(
             "upload_jobs": [
                 {
                     "job_type": "ecephys",
+                    "user_email": "abc@example.com",
                     "project_name": "Ephys Platform",
                     "platform": {
                         "name": "Electrophysiology platform",
@@ -1863,8 +1864,7 @@ def test_validate_json_v2_invalid_current(
         mock_get_job_types.return_value = ["ecephys"]
         # assume a job is already running
         job_request = SubmitJobRequestV2(
-            upload_jobs=[self.example_configs_v2],
-            user_email="abc@example.com"
+            upload_jobs=[self.example_configs_v2], user_email="abc@example.com"
         ).model_dump(mode="json", exclude_none=True)
         current_job = job_request["upload_jobs"][0]
         airflow_response = {