import logging
from typing import Optional, Union
from lxml import etree
from q2_sdk.hq.models.hq_response import HqResponse
from q2_sdk.hq.models.xsd_parser import (
DataSetSchema,
dataset_to_dict,
parse_schemas_from_element,
)
DEFAULT_LOGGER = logging.getLogger()
[docs]
class BackOfficeResponse(HqResponse):
"""
Wraps a response from HQ up as a handy object to work with.
For responses containing DataSet structures (XSD schemas + diffgram rows),
.json() will use the embedded schema to return properly-typed Python dicts
with consistent list structures for table rows.
For simple responses without DataSet structures, .json() falls back to
the generic xml_to_dict behavior.
"""
def __init__(self, raw_response: Union[str, dict], logger=None):
super().__init__(raw_response, logger=logger)
self._dataset_schemas: Optional[dict[str, DataSetSchema]] = None
self._capture_dataset_schemas()
self.clean_result_node()
self.q2_token = self.headers.get("q2token")
def _capture_dataset_schemas(self):
"""
Extract DataSet schemas from the result_node BEFORE clean_result_node()
strips them. This captured schema metadata is used by .json() for
type-aware response parsing.
"""
if self.result_node is None:
return
self._dataset_schemas = parse_schemas_from_element(self.result_node)
[docs]
def clean_result_node(self):
"""
Cleans up the grossness of the BackOffice response in a few ways:
1. Removes namespaces
2. Moves children of diffgrams to the diffgram's parent
3. Removes the diffgram node
4. Removes schema nodes
5. Objects from BackOffice typically have a parent and child node named
the same thing. Only store the child
"""
if self.result_node is None:
return
children_to_migrate = []
for elem in self.result_node.iter():
elem.tag = etree.QName(elem.tag).localname
for elem in self.result_node.iter("schema"):
elem.getparent().remove(elem)
for elem in self.result_node.iter("diffgram"):
for child in elem.getchildren():
elem.getparent().append(child)
elem.getparent().remove(elem)
for elem in self.result_node.iter():
if elem.getparent().tag == elem.tag:
children_to_migrate.append(elem)
for elem in children_to_migrate:
parent = elem.getparent()
parent.getparent().append(elem)
if not parent.getchildren():
parent.getparent().remove(parent)
[docs]
def json(self) -> dict:
"""
Convert the response to a Python dictionary.
For DataSet responses (those with embedded XSD schemas), returns a dict
with properly-typed values and consistent list structures for table rows.
For simple responses without DataSet structures, falls back to the
generic xml_to_dict parsing from HqResponse.
"""
if self.result_node is None:
raise AttributeError("result_node of BackOfficeResponse is None")
if self._dataset_schemas:
return dataset_to_dict(self.result_node, self._dataset_schemas)
# Fall back to generic xml_to_dict for simple responses
return super().json()
@staticmethod
def is_success(inner_response):
if inner_response.ErrorReturnCode.pyval != 0:
return False
return True
[docs]
@staticmethod
def get_hq_return_error(inner_response: etree.Element):
"""
Parses HQ response and returns an error message if one exists
:param inner_response: Result Node from HQ response
"""
msg = None
if inner_response.ErrorReturnCode.pyval != 0:
tag = etree.QName(inner_response.tag).localname
code = inner_response.ErrorReturnCode.pyval
msg = f"Backoffice {tag} ErrorCode: {code}"
return msg
@staticmethod
def get_error_code(inner_response: etree.Element):
return inner_response.ErrorReturnCode.pyval