|
1 | 1 | import json
|
2 | 2 | from datetime import datetime
|
3 |
| -from typing import List, Any, Awaitable |
| 3 | +from typing import List, Any, Awaitable, Optional, Dict |
4 | 4 | from time import time
|
5 | 5 | from asyncio import sleep
|
6 | 6 | from urllib.parse import urlparse, quote
|
@@ -113,25 +113,43 @@ def create_check_status_response(self, request, instance_id):
|
113 | 113 | }
|
114 | 114 | return func.HttpResponse(**response_args)
|
115 | 115 |
|
116 |
| - def get_client_response_links(self, request, instance_id): |
| 116 | + def create_http_management_payload(self, instance_id: str) -> Dict[str, str]: |
117 | 117 | """Create a dictionary of orchestrator management urls.
|
118 | 118 |
|
119 | 119 | Parameters
|
120 | 120 | ----------
|
121 |
| - request : HttpRequest |
| 121 | + instance_id : str |
| 122 | + The ID of the orchestration instance to check. |
| 123 | +
|
| 124 | + Returns |
| 125 | + ------- |
| 126 | + Dict[str, str] |
| 127 | + a dictionary object of orchestrator instance management urls |
| 128 | + """ |
| 129 | + return self.get_client_response_links(None, instance_id) |
| 130 | + |
| 131 | + def get_client_response_links( |
| 132 | + self, |
| 133 | + request: Optional[func.HttpRequest], instance_id: str) -> Dict[str, str]: |
| 134 | + """Create a dictionary of orchestrator management urls. |
| 135 | +
|
| 136 | + Parameters |
| 137 | + ---------- |
| 138 | + request : Optional[HttpRequest] |
122 | 139 | The HTTP request that triggered the current orchestration instance.
|
123 | 140 | instance_id : str
|
124 | 141 | The ID of the orchestration instance to check.
|
125 | 142 |
|
126 | 143 | Returns
|
127 | 144 | -------
|
128 |
| - dict |
| 145 | + Dict[str, str] |
129 | 146 | a dictionary object of orchestrator instance management urls
|
130 | 147 | """
|
131 | 148 | payload = self._orchestration_bindings.management_urls.copy()
|
132 | 149 |
|
133 | 150 | for key, _ in payload.items():
|
134 |
| - if request.url: |
| 151 | + request_is_not_none = not (request is None) |
| 152 | + if request_is_not_none and request.url: |
135 | 153 | payload[key] = self._replace_url_origin(request.url, payload[key])
|
136 | 154 | payload[key] = payload[key].replace(
|
137 | 155 | self._orchestration_bindings.management_urls["id"], instance_id)
|
|
0 commit comments