Identify the master cURL command responsible for the action
(self, state: AgentState)
| 225 | return state |
| 226 | |
| 227 | def url_to_curl(self, state: AgentState) -> AgentState: |
| 228 | """ |
| 229 | Identify the master cURL command responsible for the action |
| 230 | """ |
| 231 | request = self.url_to_res_req_dict[state["action_url"]]["request"] |
| 232 | curl = request.to_curl_command() |
| 233 | if curl in self.curl_to_id_dict: |
| 234 | master_node_id = self.curl_to_id_dict[curl] |
| 235 | else: |
| 236 | master_node_id = self.dag_manager.add_node( |
| 237 | node_type="master_curl", # Specify node type |
| 238 | content={ |
| 239 | "key": request, |
| 240 | "value": self.req_to_res_map[request] |
| 241 | }, |
| 242 | dynamic_parts=["None"], |
| 243 | extracted_parts=["None"] |
| 244 | ) |
| 245 | self.curl_to_id_dict[curl] = master_node_id |
| 246 | state[self.MASTER_NODE_KEY] = master_node_id |
| 247 | state[self.TO_BE_PROCESSED_NODES_KEY].append(master_node_id) |
| 248 | self.global_master_node_id = master_node_id |
| 249 | return state |
| 250 | |
| 251 | def get_simplest_request(self, request_list: List[Request]) -> Request: |
| 252 | """ |
nothing calls this directly
no test coverage detected