Unmarshal an XML-RPC response, based on incoming XML event messages (start, data, end). Call close() to get the resulting data structure. Note that this reader is fairly tolerant, and gladly accepts bogus XML-RPC data without complaining (but not bogus XML).
| 611 | # @see loads |
| 612 | |
| 613 | class Unmarshaller: |
| 614 | """Unmarshal an XML-RPC response, based on incoming XML event |
| 615 | messages (start, data, end). Call close() to get the resulting |
| 616 | data structure. |
| 617 | |
| 618 | Note that this reader is fairly tolerant, and gladly accepts bogus |
| 619 | XML-RPC data without complaining (but not bogus XML). |
| 620 | """ |
| 621 | |
| 622 | # and again, if you don't understand what's going on in here, |
| 623 | # that's perfectly ok. |
| 624 | |
| 625 | def __init__(self, use_datetime=False, use_builtin_types=False): |
| 626 | self._type = None |
| 627 | self._stack = [] |
| 628 | self._marks = [] |
| 629 | self._data = [] |
| 630 | self._value = False |
| 631 | self._methodname = None |
| 632 | self._encoding = "utf-8" |
| 633 | self.append = self._stack.append |
| 634 | self._use_datetime = use_builtin_types or use_datetime |
| 635 | self._use_bytes = use_builtin_types |
| 636 | |
| 637 | def close(self): |
| 638 | # return response tuple and target method |
| 639 | if self._type is None or self._marks: |
| 640 | raise ResponseError() |
| 641 | if self._type == "fault": |
| 642 | raise Fault(**self._stack[0]) |
| 643 | return tuple(self._stack) |
| 644 | |
| 645 | def getmethodname(self): |
| 646 | return self._methodname |
| 647 | |
| 648 | # |
| 649 | # event handlers |
| 650 | |
| 651 | def xml(self, encoding, standalone): |
| 652 | self._encoding = encoding |
| 653 | # FIXME: assert standalone == 1 ??? |
| 654 | |
| 655 | def start(self, tag, attrs): |
| 656 | # prepare to handle this element |
| 657 | if ':' in tag: |
| 658 | tag = tag.split(':')[-1] |
| 659 | if tag == "array" or tag == "struct": |
| 660 | self._marks.append(len(self._stack)) |
| 661 | self._data = [] |
| 662 | if self._value and tag not in self.dispatch: |
| 663 | raise ResponseError("unknown tag %r" % tag) |
| 664 | self._value = (tag == "value") |
| 665 | |
| 666 | def data(self, text): |
| 667 | self._data.append(text) |
| 668 | |
| 669 | def end(self, tag): |
| 670 | # call the appropriate end tag handler |
no outgoing calls
no test coverage detected
searching dependent graphs…