add disseminate by original id and type

michael-conway · michael-conway · commit 3fe492b854d9 · 2025-12-16T09:47:11.000-05:00
diff --git a/accelerator_core/service_impls/mongo_accession.py b/accelerator_core/service_impls/mongo_accession.py
@@ -131,11 +131,9 @@ def ingest(
             raise Exception("no technical metadata found, invalid record")
 
         technical_metadata["created"] = get_time_now_iso()
-
-        technical_metadata["original_source"] = (
+        technical_metadata["original_source_identifier"] = (
             ingest_payload.ingest_source_descriptor.ingest_item_id
         )
-
         technical_metadata["original_source_link"] = (
             ingest_payload.ingest_source_descriptor.ingest_link
         )
diff --git a/accelerator_core/service_impls/mongo_dissemination.py b/accelerator_core/service_impls/mongo_dissemination.py
@@ -59,7 +59,8 @@ def disseminate_by_id(
         self, document_id: str, dissemination_request: DisseminationDescriptor
     ) -> DisseminationPayload:
         """
-        Disseminate an individual document, identified by its type (parent collection) and its id
+        Disseminate an individual document, identified by its type (parent collection) and its id in the accelerator
+        database
         @param document_id: str with unique document id
         @param dissemination_request: DisseminationDescriptor that describes the type, version, and
         other information
@@ -94,6 +95,74 @@ def disseminate_by_id(
         dissemination_payload.dissemination_successful = True
         return dissemination_payload
 
+    def disseminate_by_original_source_and_id(
+        self,
+        original_source: str,
+        original_document_identifier,
+        dissemination_request: DisseminationDescriptor,
+    ) -> DisseminationPayload:
+        """
+        This method is used for creating a dissemination payload by applying a filter
+        that selects specific documents. The filtering mechanism is determined by the
+        implementation and works on provided source and document identifiers. This is meant to return a single
+        document based on its original source (e.g. "cedar") and the original identifier (e.g. the document DOI in cedar).
+
+        This is distinguised from the dissemination_by_id method which is meant to return a single document based on its
+        accelerator database identifier.
+
+        Parameters:
+            original_source: str
+                The originating source of the document, used to identify the context or
+                source system of the document.
+            original_document_identifier
+                The unique identifier of the document associated with the original source.
+            dissemination_request: DisseminationDescriptor
+                The object encapsulating metadata, rules, and instructions for how the document
+                should be disseminated.
+
+        Returns:
+            DisseminationPayload
+                The resulting payload from the dissemination operation, containing the processed
+                document and relevant dissemination metadata.
+        """
+
+        logger.info(
+            f"Disseminating document based on original source: {original_source} and document identifier: {original_document_identifier} of type {dissemination_request.ingest_type} to target: {dissemination_request.dissemination_type}"
+        )
+
+        doc = self.accel_database_utils.find_doc_by_original_source_identifier(
+            dissemination_request.ingest_type,
+            original_source,
+            original_document_identifier,
+            dissemination_request.temp_collection,
+        )
+
+        if doc is None:
+            logger.warning(
+                f"Document not found for original source: {original_source} and document identifier: {original_document_identifier} "
+            )
+            raise Exception(
+                f"Document not found for original source: {original_source} and document identifier: {original_document_identifier} "
+            )
+
+        event = create_timestamped_log(
+            f"Disseminating document  original source: {original_source} and document identifier: {original_document_identifier} of type {dissemination_request.ingest_type} to target: {dissemination_request.dissemination_type}"
+        )
+
+        """
+        self.accel_database_utils.log_document_event(
+            str(),
+            event,
+            dissemination_request.ingest_type,
+            dissemination_request.temp_collection,
+        )
+        """
+
+        dissemination_payload = DisseminationPayload(dissemination_request)
+        # self.report_individual_dissemination(dissemination_payload, document_id, doc)
+        dissemination_payload.dissemination_successful = True
+        return dissemination_payload
+
     def disseminate_by_filter(
         self,
         filter: DisseminationFilter,
diff --git a/accelerator_core/services/dissemination.py b/accelerator_core/services/dissemination.py
@@ -43,6 +43,39 @@ def disseminate_by_id(
         """
         pass
 
+    def disseminate_by_original_source_and_id(
+        self,
+        original_source: str,
+        original_document_identifier,
+        dissemination_request: DisseminationDescriptor,
+    ) -> DisseminationPayload:
+        """
+        Apply a filter to create and disseminate a single document specified target.
+
+        This method is used for creating a dissemination payload by applying a filter
+        that selects specific documents. The filtering mechanism is determined by the
+        implementation and works on provided source and document identifiers. This is meant to return a single
+        document based on its original source (e.g. "cedar") and the original identifier (e.g. the document DOI in cedar).
+
+        This is distinguised from the dissemination_by_id method which is meant to return a single document based on its
+        accelerator database identifier.
+
+        Parameters:
+            original_source: str
+                The source where the original document resides. This represents the DAG that ingested the document
+                originally
+            original_document_identifier
+                The identifier of the document in the original source. (e.g. the CEDAR generated document GUID).
+            dissemination_request: DisseminationDescriptor
+                A descriptor that details the dissemination type, version, and
+                additional relevant information.
+
+        Returns:
+            DisseminationPayload
+                The payload containing the documents selected for dissemination.
+        """
+        pass
+
     def disseminate_by_filter(
         self,
         filter: DisseminationFilter,
diff --git a/accelerator_core/workflow/accel_data_models.py b/accelerator_core/workflow/accel_data_models.py
@@ -19,10 +19,8 @@ def __init__(self):
         self.ingest_item_id = (
             None  # unique id if this is an individual item, blank for a batch
         )
-        self.ingest_link = None  # link to the ingest source
-        self.ingest_format = (
-            None  # data source of the ingest (e.g. CEDAR, CAFE, Data.Gov)
-        )
+        self.ingest_link = None  # link to the ingest source, currently is a string value indicating the original
+        # source of the data, e.g. "cedar"
         self.use_tempfiles = False  # request temp files versus inline. System behavior for a task will consult this flag
 
     def to_dict(self) -> dict:
diff --git a/integration_tests/test_dissemination_mongo.py b/integration_tests/test_dissemination_mongo.py
@@ -160,6 +160,65 @@ def test_find_one_by_filter(self):
 
             self.assertIsNotNone(actual)
 
+    def test_disseminate_by_original_source_and_id(self):
+        ingest_source_descriptor = IngestSourceDescriptor()
+        ingest_source_descriptor.ingest_type = "accelerator"
+        ingest_source_descriptor.schema_version = "1.0.2"
+        ingest_source_descriptor.ingest_identifier = "myrunid"
+        ingest_source_descriptor.ingest_item_id = (
+            "test_disseminate_by_original_source_and_id"
+        )
+        ingest_source_descriptor.ingest_link = "ingest_source"
+        ingest_source_descriptor.submitter_name = "mysubmittername"
+        ingest_source_descriptor.submitter_email = "mysubmitteremail"
+        ingest_source_descriptor.use_tempfiles = False
+
+        ingest_result = IngestPayload(ingest_source_descriptor)
+
+        json_path = determine_test_resource_path("example1.json", "integration_tests")
+        with open(json_path) as json_data:
+            d = json.load(json_data)
+            ingest_result.payload.append(d)
+            ingest_result.payload_inline = True
+
+            xcom_props_resolver = DirectXcomPropsResolver(
+                temp_files_supported=False, temp_files_location=""
+            )
+
+            accession = AccessionMongo(
+                self.__class__._accelerator_config,
+                self.__class__._accel_db_context,
+                xcom_props_resolver,
+            )
+
+            id = accession.ingest(ingest_result, check_duplicates=False, temp_doc=False)
+            self.assertIsNotNone(id)
+
+            # now get the dissemination for this item
+
+            dissemination_request = DisseminationDescriptor()
+            dissemination_request.dissemination_type = "dataverse"
+            dissemination_request.temp_collection = False
+            dissemination_request.ingest_type = "accelerator"
+            dissemination_request.schema_version = "1.0.2"
+            dissemination_request.inline_results = True
+            dissemination_request.dissemination_identifier = "test_dissemination"
+            dissemination_request.dissemination_item_id = id
+
+            dissemination = DisseminationMongo(
+                self.__class__._accelerator_config,
+                xcom_props_resolver,
+                self.__class__._accel_db_context,
+            )
+
+            dissemination_payload = dissemination.disseminate_by_original_source_and_id(
+                ingest_source_descriptor.ingest_link,
+                ingest_source_descriptor.ingest_item_id,
+                dissemination_request,
+            )
+
+            self.assertIsNotNone(dissemination_payload)
+
     def test_dissemination_not_found(self):
         ingest_source_descriptor = IngestSourceDescriptor()
         id = "95E43738404BEECDF66573B4"
diff --git a/integration_tests/test_resources/example1.json b/integration_tests/test_resources/example1.json
@@ -199,8 +199,9 @@
     "created": "",
     "modified": "",
     "verified": "",
-    "original_source_type": "",
-    "target_schema_type": "",
+    "original_source_type": "cedar",
+    "target_schema_type": "accelerator",
+    "target_schema_version": "1.0.2",
     "original_source_identifier": "",
     "original_source_link": "",
     "history": [

Original file line number	Diff line number	Diff line change
`@@ -131,11 +131,9 @@ def ingest(`
`131`	`131`	`raise Exception("no technical metadata found, invalid record")`
`132`	`132`
`133`	`133`	`technical_metadata["created"] = get_time_now_iso()`
`134`		`-`
`135`		`- technical_metadata["original_source"] = (`
	`134`	`+ technical_metadata["original_source_identifier"] = (`
`136`	`135`	`ingest_payload.ingest_source_descriptor.ingest_item_id`
`137`	`136`	`)`
`138`		`-`
`139`	`137`	`technical_metadata["original_source_link"] = (`
`140`	`138`	`ingest_payload.ingest_source_descriptor.ingest_link`
`141`	`139`	`)`