Conversation
There was a problem hiding this comment.
Pull request overview
Adds initial SDK support for the getTaskDetails task type so callers can retrieve a prior task’s original request and response and have them parsed into SDK dataclasses when possible.
Changes:
- Adds
ETaskType.GET_TASK_DETAILSplusIGetTaskDetailsRequestandITaskDetailstypes. - Implements
RunwareBase.getTaskDetails()with normalization helpers to coerce known request/response payloads into SDK dataclasses. - Documents
getTaskDetails(taskUUID)usage in the README.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| runware/types.py | Adds task type enum entry and new request/response dataclasses for task details. |
| runware/base.py | Adds getTaskDetails API call and request/response normalization logic. |
| README.md | Adds user-facing documentation and example usage for getTaskDetails. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| async def getTaskDetails(self, taskUUID: str) -> ITaskDetails: | ||
| async with self._request_semaphore: | ||
| request = IGetTaskDetailsRequest(taskUUID=taskUUID) | ||
| return await self._retry_async_with_reconnect( | ||
| self._requestTaskDetails, | ||
| request, | ||
| task_type=ETaskType.GET_TASK_DETAILS.value, | ||
| ) |
There was a problem hiding this comment.
getTaskDetails() registers the pending operation under a typed key (f"{task_uuid}:{ETaskType.GET_TASK_DETAILS.value}") to avoid collisions, but _retry_async_with_reconnect() always force-unregisters pending operations by the raw request_model.taskUUID in its finally. If getTaskDetails() is called while another operation with the same taskUUID is in-flight (stored under the raw UUID key), this call can inadvertently unregister/cancel that other operation. Consider adding an explicit pending_operation_key/unregister_key parameter to the retry helper (defaulting to taskUUID) and pass the typed key here, or otherwise ensure this call path doesn’t force-unregister the raw UUID entry.
| field_type = hints.get(k) | ||
|
|
||
| # Unwrap Optional[X] -> X | ||
| union_args = [] | ||
| if get_origin(field_type) is Union: | ||
| args = [a for a in get_args(field_type) if a is not type(None)] | ||
| field_type = args[0] if args else field_type | ||
| union_args = [a for a in get_args(field_type) if a is not type(None)] | ||
|
|
||
|
|
||
| matched_union_type = False | ||
| for arg in union_args: | ||
| if isinstance(arg, type) and isinstance(v, arg): | ||
| filtered_data[k] = v | ||
| matched_union_type = True | ||
| break | ||
| if matched_union_type: | ||
| continue | ||
|
|
||
|
|
||
| if isinstance(v, dict): | ||
| dataclass_args = [ | ||
| arg for arg in union_args | ||
| if isinstance(arg, type) and is_dataclass(arg) | ||
| ] | ||
| for dataclass_arg in dataclass_args: | ||
| try: | ||
| filtered_data[k] = instantiateDataclass(dataclass_arg, v) | ||
| matched_union_type = True | ||
| break | ||
| except Exception: | ||
| continue | ||
| if matched_union_type: | ||
| continue | ||
|
|
||
| has_dict_branch = any( | ||
| arg is dict or get_origin(arg) is dict | ||
| for arg in union_args | ||
| ) | ||
| if has_dict_branch: | ||
| filtered_data[k] = v | ||
| continue | ||
|
|
||
|
|
||
| if isinstance(v, list): | ||
| list_arg = next((arg for arg in union_args if get_origin(arg) is list), None) | ||
| if list_arg is not None: | ||
| field_type = list_arg | ||
| else: | ||
| field_type = union_args[0] if union_args else field_type | ||
| else: | ||
| field_type = union_args[0] if union_args else field_type |
There was a problem hiding this comment.
instantiateDataclass() now has significantly more logic for Union[...] fields (including dict/list branches and nested dataclass selection), but there are no unit tests covering these new Union-handling paths. Adding focused tests (e.g., Optional[Union[SomeDataclass, Dict[str, Any]]], Optional[Union[List[SomeDataclass], List[Dict[str, Any]]]], and ensuring dict branches remain dicts) would help prevent regressions since this utility is used to parse API responses across the SDK.
Added
getTaskDetails