Use existing assoc. table for OP to DO

Author: naglepuff

nmdc_server/aggregations.py

@@ -234,10 +234,15 @@ def get_data_object_aggregation(
             func.count(models.DataObject.id),
             func.sum(func.coalesce(models.DataObject.file_size_bytes, 0)),
         )
+        .join(
+            models.omics_processing_output_association,
+            models.omics_processing_output_association.c.data_object_id
+            == models.DataObject.id
+        )
         .filter(
             models.DataObject.workflow_type != None,
             models.DataObject.file_type != None,
-            subquery.c.id == models.DataObject.omics_processing_id,
+            subquery.c.id == models.omics_processing_output_association.c.omics_processing_id,
             models.DataObject.url != None,
         )
         .group_by(models.DataObject.workflow_type, models.DataObject.file_type)
@@ -255,9 +260,14 @@ def get_data_object_aggregation(
             func.count(models.DataObject.id),
             func.sum(func.coalesce(models.DataObject.file_size_bytes, 0)),
         )
+        .join(
+            models.omics_processing_output_association,
+            models.omics_processing_output_association.c.data_object_id
+            == models.DataObject.id
+        )
         .filter(
             models.DataObject.workflow_type != None,
-            subquery.c.id == models.DataObject.omics_processing_id,
+            subquery.c.id == models.omics_processing_output_association.c.omics_processing_id,
             models.DataObject.url != None,
         )
         .group_by(models.DataObject.workflow_type)
@@ -274,10 +284,15 @@ def get_data_object_aggregation(
             func.count(models.DataObject.id),
             func.sum(func.coalesce(models.DataObject.file_size_bytes, 0)),
         )
+        .join(
+            models.omics_processing_output_association,
+            models.omics_processing_output_association.c.data_object_id
+            == models.DataObject.id
+        )
         .filter(
             models.DataObject.workflow_type != None,
             models.DataObject.file_type != None,
-            subquery.c.id == models.DataObject.omics_processing_id,
+            subquery.c.id == models.omics_processing_output_association.c.omics_processing_id,
             models.DataObject.url != None,
         )
         .group_by(models.DataObject.workflow_type, models.DataObject.file_type)

nmdc_server/ingest/all.py

@@ -101,14 +101,6 @@ def load(db: Session, function_limit=None, skip_annotation=False):
     )
     db.commit()
 
-    # Update the FK relationship from the data_object table to the
-    # omics_processing table.
-    logger.info("Updating foreign key relationship from Data Object to Data Generation")
-    data_object.update_data_generation_relation(
-        db,
-        mongodb["data_object_set"].find(),
-    )
-
     workflow_set = "workflow_execution_set"
 
     logger.info("Loading metabolomics analysis...")

nmdc_server/ingest/data_object.py

@@ -43,23 +43,3 @@ def load(db: Session, cursor: Cursor, file_types: List[Dict[str, Any]]):
 
     if objects_without_type:
         logger.error(f"Encountered {objects_without_type} objects without data_object_type")
-
-
-def update_data_generation_relation(db: Session, cursor: Cursor):
-    """
-    Update DataObject's omics_processing_id FK.
-
-    This should run after ingesting all data objects and data generations (omics processing).
-    """
-    for data_object in cursor:
-        id = data_object["id"]
-        was_generated_by = data_object.pop("was_generated_by", None)
-        if not was_generated_by:
-            continue
-        # Mypy does not like db.get, and reports that "Session" has no attribute "get."
-        # See https://docs.sqlalchemy.org/en/14/orm/session_basics.html#get-by-primary-key
-        data_generation = db.get(OmicsProcessing, was_generated_by)  # type: ignore
-        row = db.get(DataObject, id)  # type: ignore
-        if row and data_generation:
-            row.omics_processing_id = was_generated_by
-            db.add(row)

nmdc_server/ingest/pipeline.py

@@ -308,6 +308,15 @@ def load(
                 .on_conflict_do_nothing()
             )
 
+
+            for data_generation in was_informed_by:
+                data_objects = inputs + outputs
+                db.execute(
+                    insert(models.omics_processing_output_association)
+                    .values([(data_generation, data_object) for data_object in data_objects])
+                    .on_conflict_do_nothing()
+                )
+
         for id_ in outputs:
             output = db.query(models.DataObject).get(id_)
             assert output

nmdc_server/models.py

@@ -819,6 +819,8 @@ class DataObject(Base):
     workflow_type = Column(String, nullable=True)
 
     # denormalized relationship representing the source omics_processing
+    # TODO: investigate whether or not these can be removed completely in
+    # favor of the association table omics_processing_output_association
     omics_processing_id = Column(String, ForeignKey("omics_processing.id"), nullable=True)
     omics_processing = relationship(OmicsProcessing)
 

nmdc_server/query.py

@@ -963,8 +963,12 @@ def _data_object_filter_subquery(
     ) -> Query:
         """Create a subquery that selects from a data object filter condition."""
         query = db.query(models.DataObject.id.label("id")).join(
+            models.omics_processing_output_association,
+            models.omics_processing_output_association.c.id
+            == models.DataObject.id
+        ).join(
             op_cte,
-            models.DataObject.omics_processing_id == op_cte.c.id,
+            models.omics_processing_output_association.c.omics_processing_id == op_cte.c.id,
         )
         if filter.workflow:
             query = query.filter(models.DataObject.workflow_type == filter.workflow.value)