Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.datatypes.dataproviders.hierarchy
"""
Dataproviders that iterate over lines from their sources.
"""
import logging
from galaxy.util import (
Element,
etree,
)
from . import line
_TODO = """
"""
log = logging.getLogger(__name__)
# ----------------------------------------------------------------------------- hierarchal/tree data providers
[docs]class HierarchalDataProvider(line.BlockDataProvider):
"""
Class that uses formats where a datum may have a parent or children
data.
e.g. XML, HTML, GFF3, Phylogenetic
"""
[docs] def __init__(self, source, **kwargs):
# TODO: (and defer to better (than I can write) parsers for each subtype)
super().__init__(source, **kwargs)
# ----------------------------------------------------------------------------- xml
[docs]class XMLDataProvider(HierarchalDataProvider):
"""
Data provider that converts selected XML elements to dictionaries.
"""
# using lxml.etree's iterparse method to keep mem down
# TODO: this, however (AFAIK), prevents the use of xpath
settings = {
"selector": "str", # urlencoded
"max_depth": "int",
}
ITERPARSE_ALL_EVENTS = ("start", "end", "start-ns", "end-ns")
# TODO: move appropo into super
[docs] def __init__(self, source, selector=None, max_depth=None, **kwargs):
"""
:param selector: some partial string in the desired tags to return
:param max_depth: the number of generations of descendents to return
"""
self.selector = selector
self.max_depth = max_depth
self.namespaces = {}
super().__init__(source, **kwargs)
[docs] def matches_selector(self, element, selector=None):
"""
Returns true if the ``element`` matches the ``selector``.
:param element: an XML ``Element``
:param selector: some partial string in the desired tags to return
Change point for more sophisticated selectors.
"""
# search for partial match of selector to the element tag
# TODO: add more flexibility here w/o re-implementing xpath
# TODO: fails with '#' - browser thinks it's an anchor - use urlencode
# TODO: need removal/replacement of etree namespacing here - then move to string match
return bool((selector is None) or (isinstance(element, Element) and selector in element.tag))
[docs] def element_as_dict(self, element):
"""
Converts an XML element (its text, tag, and attributes) to dictionary form.
:param element: an XML ``Element``
"""
# TODO: Key collision is unlikely here, but still should be better handled
return {
"tag": element.tag,
"text": element.text.strip() if element.text else None,
# needs shallow copy to protect v. element.clear()
"attrib": dict(element.attrib),
}
[docs] def get_children(self, element, max_depth=None):
"""
Yield all children of element (and their children - recursively)
in dictionary form.
:param element: an XML ``Element``
:param max_depth: the number of generations of descendents to return
"""
if not isinstance(max_depth, int) or max_depth >= 1:
for child in element:
child_data = self.element_as_dict(child)
next_depth = max_depth - 1 if isinstance(max_depth, int) else None
grand_children = list(self.get_children(child, next_depth))
if grand_children:
child_data["children"] = grand_children
yield child_data
def __iter__(self):
context = etree.iterparse(self.source, events=self.ITERPARSE_ALL_EVENTS)
context = iter(context)
selected_element = None
for event, element in context:
if event == "start-ns":
ns, uri = element
self.namespaces[ns] = uri
elif event == "start":
if (selected_element is None) and (self.matches_selector(element, self.selector)):
# start tag of selected element - wait for 'end' to emit/yield
selected_element = element
elif event == "end":
if (selected_element is not None) and (element == selected_element):
self.num_valid_data_read += 1
# offset
if self.num_valid_data_read > self.offset:
# convert to dict and yield
selected_element_dict = self.element_as_dict(selected_element)
children = list(self.get_children(selected_element, self.max_depth))
if children:
selected_element_dict["children"] = children
yield selected_element_dict
# limit
self.num_data_returned += 1
if self.limit is not None and self.num_data_returned >= self.limit:
break
selected_element.clear()
selected_element = None
self.num_data_read += 1