|
2 | 2 |
|
3 | 3 | from collections.abc import Mapping |
4 | 4 | from typing import Any |
5 | | -from xml.parsers.expat import ExpatError |
6 | 5 |
|
7 | 6 | import xmltodict |
8 | 7 |
|
@@ -72,7 +71,7 @@ def publish(self, path: str, files: Mapping[str, Any] | None) -> int: |
72 | 71 | If the server returns an error during upload. |
73 | 72 | """ |
74 | 73 | response = self._http.post(path, files=files) |
75 | | - parsed_response = self._parse_xml_response(response.content) |
| 74 | + parsed_response = xmltodict.parse(response.content) |
76 | 75 | return self._extract_id_from_upload(parsed_response) |
77 | 76 |
|
78 | 77 | def delete(self, resource_id: int) -> bool: |
@@ -107,7 +106,7 @@ def delete(self, resource_id: int) -> bool: |
107 | 106 | path = f"{endpoint_name}/{resource_id}" |
108 | 107 | try: |
109 | 108 | response = self._http.delete(path) |
110 | | - result = self._parse_xml_response(response.content) |
| 109 | + result = xmltodict.parse(response.content) |
111 | 110 | return f"oml:{endpoint_name}_delete" in result |
112 | 111 | except OpenMLServerException as e: |
113 | 112 | self._handle_delete_exception(endpoint_name, e) |
@@ -144,7 +143,7 @@ def tag(self, resource_id: int, tag: str) -> list[str]: |
144 | 143 | data = {f"{endpoint_name}_id": resource_id, "tag": tag} |
145 | 144 | response = self._http.post(path, data=data) |
146 | 145 |
|
147 | | - parsed_response = self._parse_xml_response(response.content, force_list={"oml:tag"}) |
| 146 | + parsed_response = xmltodict.parse(response.content, force_list={"oml:tag"}) |
148 | 147 | result = parsed_response[f"oml:{endpoint_name}_tag"] |
149 | 148 | tags: list[str] = result.get("oml:tag", []) |
150 | 149 |
|
@@ -181,30 +180,12 @@ def untag(self, resource_id: int, tag: str) -> list[str]: |
181 | 180 | data = {f"{endpoint_name}_id": resource_id, "tag": tag} |
182 | 181 | response = self._http.post(path, data=data) |
183 | 182 |
|
184 | | - parsed_response = self._parse_xml_response(response.content, force_list={"oml:tag"}) |
| 183 | + parsed_response = xmltodict.parse(response.content, force_list={"oml:tag"}) |
185 | 184 | result = parsed_response[f"oml:{endpoint_name}_untag"] |
186 | 185 | tags: list[str] = result.get("oml:tag", []) |
187 | 186 |
|
188 | 187 | return tags |
189 | 188 |
|
190 | | - def _parse_xml_response(self, payload: bytes | str, **kwargs: Any) -> Mapping[str, Any]: |
191 | | - try: |
192 | | - parsed_response: Mapping[str, Any] = xmltodict.parse(payload, **kwargs) |
193 | | - return parsed_response |
194 | | - except ExpatError: |
195 | | - payload_text = ( |
196 | | - payload.decode("utf-8", errors="ignore") if isinstance(payload, bytes) else payload |
197 | | - ) |
198 | | - xml_start = payload_text.find("<?xml") |
199 | | - if xml_start == -1: |
200 | | - xml_start = payload_text.find("<oml:") |
201 | | - if xml_start == -1: |
202 | | - raise |
203 | | - |
204 | | - xml_text = payload_text[xml_start:] |
205 | | - parsed_fallback: Mapping[str, Any] = xmltodict.parse(xml_text, **kwargs) |
206 | | - return parsed_fallback |
207 | | - |
208 | 189 | def _get_endpoint_name(self) -> str: |
209 | 190 | if self.resource_type == ResourceType.DATASET: |
210 | 191 | return "data" |
|
0 commit comments