mage/python/xml_module.py at main · memgraph/mage · GitHub

This repository was archived by the owner on Jan 23, 2026. It is now read-only.

History

159 lines (134 loc) · 4.76 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

import mgp

import defusedxml.ElementTree as ET

import urllib.request

TYPE = "_type"

TEXT = "_text"

CHILDREN = "_children"

def parse_element(element, simple):

result = {TYPE: element.tag}

attributes = element.attrib

if attributes:

result.update(attributes)

text_content = element.text

if text_content and text_content.strip():

result[TEXT] = text_content

children = list(element)

if children:

children_name = CHILDREN

if simple:

children_name = "_" + str(element.tag)

result[children_name] = [parse_element(child, simple) for child in children]

return result

def xml_file_to_string(xml_file):

xml_string = ""

try:

with open(xml_file, "r") as xml_file:

xml_string = xml_file.read().replace("\n", "").replace(" ", "")

except PermissionError:

raise PermissionError(

"You don't have permissions to write into that file."

"Make sure to give the necessary permissions to user memgraph."

)

except Exception:

raise OSError("Could not open or write to file.")

return xml_string

@mgp.function

def parse(xml_input: str, simple: bool = False, path: str = "") -> mgp.Map:

"""

Function to parse xml string (or file) into a map.

Parameters

----------

xml_input : str

XML string which is to be parsed.

simple: bool = false

Boolean which specifies how should the children list be named,

when it is false, all children lists are named _children

if true, all children lists are named based on their parent.

path: str = ""

Path to XML file which is to be parsed, if it is not "",

XML string is ignored and only file is parsed, if it is left as

default, it is ignored.

Returns:

mgp.Map -> XML file parsed as map

"""

root = None

parser = ET.DefusedXMLParser()

if path:

root = ET.fromstring(xml_file_to_string(path), parser)

else:

root = ET.fromstring(xml_input, parser)

output_map = parse_element(root, simple)

return output_map

def check_url(url):

if not url.endswith(".xml"):

raise ValueError("File must be xml!")

def xpath_search(root, xpath_expression):

try:

result = root.findall(xpath_expression)

return result

except Exception as e:

raise ValueError(f"XPath search error: {e}")

"""

Note: our implementation of xpath is different from neo4j,

because python offers different support for xpath than java

We cant have absolute paths

And our xpath search starts from the root node,

so ./something is equivalent to /root/something

For example,

python input : .//ZONE is equivalent to java input //ZONE

https://docs.python.org/3/library/xml.etree.elementtree.html#xpath-support

here are our avaliable xpath options

"""

@mgp.read_proc

def load(

xml_url: str,

simple: bool = False,

path: str = "",

xpath: str = "",

headers: mgp.Map = {},

) -> mgp.Record(output_map=mgp.Map):

"""

Procedure to load XML from url or from file to a map.

Parameters

----------

xml_url : str

Url of the xml file to be parsed.

simple: bool = false

Boolean which specifies how should the children list be named,

when it is false, all children lists are named _children

if true, all children lists are named based on their parent.

path: str = ""

Path to XML file which is to be parsed, if it is not "",

XML url is ignored and only file is parsed. If it is left as default,

file path is ignored.

xpath: str = ""

Xpath expression which specifies which elements shall be returned.

If left as "", it will be ignored, otherwise,

only elements in XML file which satisfy

expression are returned.

headers: mgp.Map ={}

Additional HTTP headers used in url request.

Returns:

mgp.Map -> XML file or URL parsed as map.

In case XPATH is active, a map of for each element will be returned.

"""

root = None

parser = ET.DefusedXMLParser()

if path:

root = ET.fromstring(xml_file_to_string(path), parser)

else:

check_url(xml_url)

try:

request = urllib.request.Request(xml_url, headers=headers)

response = urllib.request.urlopen(request).read()

root = ET.fromstring(response)

except Exception as e:

raise ValueError(f"Error while fetching or parsing XML: {e}")

if xpath:

record_list = list()

xpath_list = xpath_search(root, xpath)

for element in xpath_list:

record_list.append(mgp.Record(output_map=parse_element(element, simple)))

return record_list

output_map = parse_element(root, simple)

return mgp.Record(output_map=output_map)

mage/python/xml_module.py at main · memgraph/mage · GitHub

Expand file tree

Breadcrumbs

xml_module.py

Latest commit

History

Breadcrumbs

xml_module.py

File metadata and controls