Merge pull request galaxyproject#17958 from jdavcs/dev_model_edits2

Model typing and SA2.0 follow-up

lib/galaxy/jobs/__init__.py

@@ -1554,7 +1554,7 @@ def change_state(self, state, info=False, flush=True, job=None):
     def get_state(self) -> str:
         job = self.get_job()
         self.sa_session.refresh(job)
-        return job.state  # type:ignore[return-value]
+        return job.state
 
     def set_runner(self, runner_url, external_id):
         log.warning("set_runner() is deprecated, use set_job_destination()")

lib/galaxy/managers/jobs.py

@@ -5,7 +5,10 @@
     date,
     datetime,
 )
-from typing import List
+from typing import (
+    cast,
+    List,
+)
 
 import sqlalchemy
 from boltons.iterutils import remap
@@ -1065,6 +1068,7 @@ def summarize_job_outputs(job: model.Job, tool, params):
         ("hdca", "dataset_collection_id", job.output_dataset_collection_instances),
     )
     for src, attribute, output_associations in possible_outputs:
+        output_associations = cast(List, output_associations)  # during iteration, mypy sees it as object
         for output_association in output_associations:
             output_name = output_association.name
             if output_name not in output_labels and tool:

lib/galaxy/managers/notification.py

@@ -275,8 +275,8 @@ def update_broadcasted_notification(self, notification_id: int, request: Notific
     def get_user_notification_preferences(self, user: User) -> UserNotificationPreferences:
         """Gets the user's current notification preferences or the default ones if no preferences are set."""
         current_notification_preferences = (
-            user.preferences[NOTIFICATION_PREFERENCES_SECTION_NAME]  # type:ignore[index]
-            if NOTIFICATION_PREFERENCES_SECTION_NAME in user.preferences  # type:ignore[operator]
+            user.preferences[NOTIFICATION_PREFERENCES_SECTION_NAME]
+            if NOTIFICATION_PREFERENCES_SECTION_NAME in user.preferences
             else None
         )
         try:
@@ -291,9 +291,7 @@ def update_user_notification_preferences(
         """Updates the user's notification preferences with the requested changes."""
         notification_preferences = self.get_user_notification_preferences(user)
         notification_preferences.update(request.preferences)
-        user.preferences[NOTIFICATION_PREFERENCES_SECTION_NAME] = (
-            notification_preferences.model_dump_json()
-        )  # type:ignore[index]
+        user.preferences[NOTIFICATION_PREFERENCES_SECTION_NAME] = notification_preferences.model_dump_json()
         with transaction(self.sa_session):
             self.sa_session.commit()
         return notification_preferences

lib/galaxy/managers/sharable.py

@@ -179,7 +179,7 @@ def share_with(self, item, user: User, flush: bool = True):
         """
         # precondition: user has been validated
         # get or create
-        existing = self.get_share_assocs(item, user=user)  # type:ignore[dict-item]
+        existing = self.get_share_assocs(item, user=user)
         if existing:
             return existing.pop(0)
         return self._create_user_share_assoc(item, user, flush=flush)

lib/galaxy/model/store/__init__.py

@@ -2065,78 +2065,84 @@ def export_jobs(
                 output_dataset_collection_mapping: Dict[str, List[Union[str, int]]] = {}
                 implicit_output_dataset_collection_mapping: Dict[str, List[Union[str, int]]] = {}
 
-                for assoc in job.input_datasets:
+                for id_assoc in job.input_datasets:
                     # Optional data inputs will not have a dataset.
-                    if assoc.dataset:
-                        name = assoc.name
+                    if id_assoc.dataset:
+                        name = id_assoc.name
                         if name not in input_dataset_mapping:
                             input_dataset_mapping[name] = []
 
-                        input_dataset_mapping[name].append(self.exported_key(assoc.dataset))
+                        input_dataset_mapping[name].append(self.exported_key(id_assoc.dataset))
                         if include_job_data:
-                            self.add_dataset(assoc.dataset)
+                            self.add_dataset(id_assoc.dataset)
 
-                for assoc in job.output_datasets:
+                for od_assoc in job.output_datasets:
                     # Optional data inputs will not have a dataset.
-                    if assoc.dataset:
-                        name = assoc.name
+                    if od_assoc.dataset:
+                        name = od_assoc.name
                         if name not in output_dataset_mapping:
                             output_dataset_mapping[name] = []
 
-                        output_dataset_mapping[name].append(self.exported_key(assoc.dataset))
+                        output_dataset_mapping[name].append(self.exported_key(od_assoc.dataset))
                         if include_job_data:
-                            self.add_dataset(assoc.dataset)
+                            self.add_dataset(od_assoc.dataset)
 
-                for assoc in job.input_dataset_collections:
+                for idc_assoc in job.input_dataset_collections:
                     # Optional data inputs will not have a dataset.
-                    if assoc.dataset_collection:
-                        name = assoc.name
+                    if idc_assoc.dataset_collection:
+                        name = idc_assoc.name
                         if name not in input_dataset_collection_mapping:
                             input_dataset_collection_mapping[name] = []
 
-                        input_dataset_collection_mapping[name].append(self.exported_key(assoc.dataset_collection))
+                        input_dataset_collection_mapping[name].append(self.exported_key(idc_assoc.dataset_collection))
                         if include_job_data:
-                            self.export_collection(assoc.dataset_collection)
+                            self.export_collection(idc_assoc.dataset_collection)
 
-                for assoc in job.input_dataset_collection_elements:
-                    if assoc.dataset_collection_element:
-                        name = assoc.name
+                for idce_assoc in job.input_dataset_collection_elements:
+                    if idce_assoc.dataset_collection_element:
+                        name = idce_assoc.name
                         if name not in input_dataset_collection_element_mapping:
                             input_dataset_collection_element_mapping[name] = []
 
                         input_dataset_collection_element_mapping[name].append(
-                            self.exported_key(assoc.dataset_collection_element)
+                            self.exported_key(idce_assoc.dataset_collection_element)
                         )
                         if include_job_data:
-                            if assoc.dataset_collection_element.is_collection:
-                                self.export_collection(assoc.dataset_collection_element.element_object)
+                            if idce_assoc.dataset_collection_element.is_collection:
+                                assert isinstance(
+                                    idce_assoc.dataset_collection_element.element_object, model.DatasetCollection
+                                )
+                                self.export_collection(idce_assoc.dataset_collection_element.element_object)
                             else:
-                                self.add_dataset(assoc.dataset_collection_element.element_object)
+                                assert isinstance(
+                                    idce_assoc.dataset_collection_element.element_object, model.DatasetInstance
+                                )
+                                self.add_dataset(idce_assoc.dataset_collection_element.element_object)
 
-                for assoc in job.output_dataset_collection_instances:
+                for odci_assoc in job.output_dataset_collection_instances:
                     # Optional data outputs will not have a dataset.
                     # These are implicit outputs, we don't need to export them
-                    if assoc.dataset_collection_instance:
-                        name = assoc.name
+                    if odci_assoc.dataset_collection_instance:
+                        name = odci_assoc.name
                         if name not in output_dataset_collection_mapping:
                             output_dataset_collection_mapping[name] = []
 
                         output_dataset_collection_mapping[name].append(
-                            self.exported_key(assoc.dataset_collection_instance)
+                            self.exported_key(odci_assoc.dataset_collection_instance)
                         )
 
-                for assoc in job.output_dataset_collections:
-                    if assoc.dataset_collection:
-                        name = assoc.name
+                for odc_assoc in job.output_dataset_collections:
+                    if odc_assoc.dataset_collection:
+                        name = odc_assoc.name
 
                         if name not in implicit_output_dataset_collection_mapping:
                             implicit_output_dataset_collection_mapping[name] = []
 
                         implicit_output_dataset_collection_mapping[name].append(
-                            self.exported_key(assoc.dataset_collection)
+                            self.exported_key(odc_assoc.dataset_collection)
                         )
                         if include_job_data:
-                            self.export_collection(assoc.dataset_collection)
+                            self.export_collection(odc_assoc.dataset_collection)
 
                 job_attrs["input_dataset_mapping"] = input_dataset_mapping
                 job_attrs["input_dataset_collection_mapping"] = input_dataset_collection_mapping

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -264,7 +264,7 @@ def common_problems(
         for job_input_assoc in job.input_datasets:
             input_dataset_instance = job_input_assoc.dataset
             if input_dataset_instance is None:
-                continue
+                continue  # type:ignore[unreachable]  # TODO if job_input_assoc.dataset is indeed never None, remove the above check
             if input_dataset_instance.get_total_size() == 0:
                 has_empty_inputs = True
             input_instance_id = input_dataset_instance.id

lib/galaxy/workflow/modules.py

@@ -2516,13 +2516,13 @@ def inject(self, step: WorkflowStep, step_args=None, steps=None, **kwargs):
         If step_args is provided from a web form this is applied to generate
         'state' else it is just obtained from the database.
         """
-        step.upgrade_messages = {}  # type: ignore[assignment]
+        step.upgrade_messages = {}
 
         # Make connection information available on each step by input name.
         step.setup_input_connections_by_name()
 
         # Populate module.
-        module = step.module = module_factory.from_workflow_step(self.trans, step, **kwargs)  # type: ignore[assignment]
+        module = step.module = module_factory.from_workflow_step(self.trans, step, **kwargs)
 
         # Any connected input needs to have value DummyDataset (these
         # are not persisted so we need to do it every time)