CreateHttpManagementPayload

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -1,6 +1,6 @@
 import json
 from datetime import datetime
-from typing import List, Any, Awaitable
+from typing import List, Any, Awaitable, Optional, Dict
 from time import time
 from asyncio import sleep
 from urllib.parse import urlparse, quote
@@ -113,25 +113,43 @@ def create_check_status_response(self, request, instance_id):
         }
         return func.HttpResponse(**response_args)
 
-    def get_client_response_links(self, request, instance_id):
+    def create_http_management_payload(self, instance_id: str) -> Dict[str, str]:
         """Create a dictionary of orchestrator management urls.
 
         Parameters
         ----------
-        request : HttpRequest
+        instance_id : str
+            The ID of the orchestration instance to check.
+
+        Returns
+        -------
+        Dict[str, str]
+            a dictionary object of orchestrator instance management urls
+        """
+        return self.get_client_response_links(None, instance_id)
+
+    def get_client_response_links(
+            self,
+            request: Optional[func.HttpRequest], instance_id: str) -> Dict[str, str]:
+        """Create a dictionary of orchestrator management urls.
+
+        Parameters
+        ----------
+        request : Optional[HttpRequest]
             The HTTP request that triggered the current orchestration instance.
         instance_id : str
             The ID of the orchestration instance to check.
 
         Returns
         -------
-        dict
+        Dict[str, str]
             a dictionary object of orchestrator instance management urls
         """
         payload = self._orchestration_bindings.management_urls.copy()
 
         for key, _ in payload.items():
-            if request.url:
+            request_is_not_none = not (request is None)
+            if request_is_not_none and request.url:
                 payload[key] = self._replace_url_origin(request.url, payload[key])
             payload[key] = payload[key].replace(
                 self._orchestration_bindings.management_urls["id"], instance_id)