feat(rest-emitter): adding async flag to rest emitter (#10902)

Co-authored-by: Gabe Lyons <[email protected]>
datahub-project · Jul 12, 2024 · 423af83 · 423af83
1 parent 542f6c1
commit 423af83
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 6 deletions.
diff --git a/metadata-ingestion/src/datahub/emitter/rest_emitter.py b/metadata-ingestion/src/datahub/emitter/rest_emitter.py
@@ -206,14 +206,15 @@ def emit(
             UsageAggregation,
         ],
         callback: Optional[Callable[[Exception, str], None]] = None,
+        async_flag: Optional[bool] = None,
     ) -> None:
         try:
             if isinstance(item, UsageAggregation):
                 self.emit_usage(item)
             elif isinstance(
                 item, (MetadataChangeProposal, MetadataChangeProposalWrapper)
             ):
-                self.emit_mcp(item)
+                self.emit_mcp(item, async_flag=async_flag)
             else:
                 self.emit_mce(item)
         except Exception as e:
@@ -245,24 +246,39 @@ def emit_mce(self, mce: MetadataChangeEvent) -> None:
         self._emit_generic(url, payload)
 
     def emit_mcp(
-        self, mcp: Union[MetadataChangeProposal, MetadataChangeProposalWrapper]
+        self,
+        mcp: Union[MetadataChangeProposal, MetadataChangeProposalWrapper],
+        async_flag: Optional[bool] = None,
     ) -> None:
         url = f"{self._gms_server}/aspects?action=ingestProposal"
         ensure_has_system_metadata(mcp)
+
         mcp_obj = pre_json_transform(mcp.to_obj())
-        payload = json.dumps({"proposal": mcp_obj})
+        payload_dict = {"proposal": mcp_obj}
+
+        if async_flag is not None:
+            payload_dict["async"] = "true" if async_flag else "false"
+
+        payload = json.dumps(payload_dict)
 
         self._emit_generic(url, payload)
 
     def emit_mcps(
-        self, mcps: List[Union[MetadataChangeProposal, MetadataChangeProposalWrapper]]
+        self,
+        mcps: List[Union[MetadataChangeProposal, MetadataChangeProposalWrapper]],
+        async_flag: Optional[bool] = None,
     ) -> None:
         url = f"{self._gms_server}/aspects?action=ingestProposalBatch"
         for mcp in mcps:
             ensure_has_system_metadata(mcp)
 
         mcp_objs = [pre_json_transform(mcp.to_obj()) for mcp in mcps]
-        payload = json.dumps({"proposals": mcp_objs})
+        payload_dict: dict = {"proposals": mcp_objs}
+
+        if async_flag is not None:
+            payload_dict["async"] = "true" if async_flag else "false"
+
+        payload = json.dumps(payload_dict)
 
         self._emit_generic(url, payload)
 

diff --git a/metadata-ingestion/tests/test_helpers/graph_helpers.py b/metadata-ingestion/tests/test_helpers/graph_helpers.py
@@ -111,14 +111,17 @@ def emit(
             UsageAggregationClass,
         ],
         callback: Union[Callable[[Exception, str], None], None] = None,
+        async_flag: Optional[bool] = None,
     ) -> None:
         self.emitted.append(item)  # type: ignore
 
     def emit_mce(self, mce: MetadataChangeEvent) -> None:
         self.emitted.append(mce)
 
     def emit_mcp(
-        self, mcp: Union[MetadataChangeProposal, MetadataChangeProposalWrapper]
+        self,
+        mcp: Union[MetadataChangeProposal, MetadataChangeProposalWrapper],
+        async_flag: Optional[bool] = None,
     ) -> None:
         self.emitted.append(mcp)