Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/rdflib/plugins/serializers/xmlwriter.py @ 1:56ad4e20f292 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
comparison
equal deleted inserted replaced
0:d30785e31577 1:56ad4e20f292
1 import codecs
2 from xml.sax.saxutils import quoteattr, escape
3
4 __all__ = ['XMLWriter']
5
6 ESCAPE_ENTITIES = {
7 '\r': '
'
8 }
9
10
11 class XMLWriter(object):
12 def __init__(self, stream, namespace_manager, encoding=None,
13 decl=1, extra_ns=None):
14 encoding = encoding or 'utf-8'
15 encoder, decoder, stream_reader, stream_writer = \
16 codecs.lookup(encoding)
17 self.stream = stream = stream_writer(stream)
18 if decl:
19 stream.write('<?xml version="1.0" encoding="%s"?>' % encoding)
20 self.element_stack = []
21 self.nm = namespace_manager
22 self.extra_ns = extra_ns or {}
23 self.closed = True
24
25 def __get_indent(self):
26 return " " * len(self.element_stack)
27 indent = property(__get_indent)
28
29 def __close_start_tag(self):
30 if not self.closed: # TODO:
31 self.closed = True
32 self.stream.write(">")
33
34 def push(self, uri):
35 self.__close_start_tag()
36 write = self.stream.write
37 write("\n")
38 write(self.indent)
39 write("<%s" % self.qname(uri))
40 self.element_stack.append(uri)
41 self.closed = False
42 self.parent = False
43
44 def pop(self, uri=None):
45 top = self.element_stack.pop()
46 if uri:
47 assert uri == top
48 write = self.stream.write
49 if not self.closed:
50 self.closed = True
51 write("/>")
52 else:
53 if self.parent:
54 write("\n")
55 write(self.indent)
56 write("</%s>" % self.qname(top))
57 self.parent = True
58
59 def element(self, uri, content, attributes={}):
60 """Utility method for adding a complete simple element"""
61 self.push(uri)
62 for k, v in attributes.items():
63 self.attribute(k, v)
64 self.text(content)
65 self.pop()
66
67 def namespaces(self, namespaces=None):
68 if not namespaces:
69 namespaces = self.nm.namespaces()
70
71 write = self.stream.write
72 write("\n")
73 for prefix, namespace in namespaces:
74 if prefix:
75 write(' xmlns:%s="%s"\n' % (prefix, namespace))
76 # Allow user-provided namespace bindings to prevail
77 elif prefix not in self.extra_ns:
78 write(' xmlns="%s"\n' % namespace)
79
80 for prefix, namespace in list(self.extra_ns.items()):
81 if prefix:
82 write(' xmlns:%s="%s"\n' % (prefix, namespace))
83 else:
84 write(' xmlns="%s"\n' % namespace)
85
86 def attribute(self, uri, value):
87 write = self.stream.write
88 write(" %s=%s" % (self.qname(uri), quoteattr(value)))
89
90 def text(self, text):
91 self.__close_start_tag()
92 if "<" in text and ">" in text and not "]]>" in text:
93 self.stream.write("<![CDATA[")
94 self.stream.write(text)
95 self.stream.write("]]>")
96 else:
97 self.stream.write(escape(text, ESCAPE_ENTITIES))
98
99 def qname(self, uri):
100 """Compute qname for a uri using our extra namespaces,
101 or the given namespace manager"""
102
103 for pre, ns in list(self.extra_ns.items()):
104 if uri.startswith(ns):
105 if pre != "":
106 return ":".join(pre, uri[len(ns):])
107 else:
108 return uri[len(ns):]
109
110 return self.nm.qname(uri)