Statistics
| Branch: | Tag: | Revision:

root / ncclient / content.py @ 4de03d63

History | View | Annotate | Download (4.7 kB)

1
# Copyright 2009 Shikhar Bhushan
2
#
3
# Licensed under the Apache License, Version 2.0 (the "License");
4
# you may not use this file except in compliance with the License.
5
# You may obtain a copy of the License at
6
#
7
#    http://www.apache.org/licenses/LICENSE-2.0
8
#
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS,
11
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
# See the License for the specific language governing permissions and
13
# limitations under the License.
14

    
15
from cStringIO import StringIO
16
from xml.etree import cElementTree as ET
17

    
18
from ncclient import NCClientError
19

    
20
class ContentError(NCClientError):
21
    pass
22

    
23
### Namespace-related
24

    
25
BASE_NS = 'urn:ietf:params:xml:ns:netconf:base:1.0'
26
# and this is BASE_NS according to cisco devices...
27
CISCO_BS = 'urn:ietf:params:netconf:base:1.0'
28

    
29
try:
30
    register_namespace = ET.register_namespace
31
except AttributeError:
32
    def register_namespace(prefix, uri):
33
        from xml.etree import ElementTree
34
        # cElementTree uses ElementTree's _namespace_map, so that's ok
35
        ElementTree._namespace_map[uri] = prefix
36

    
37
# we'd like BASE_NS to be prefixed as "netconf"
38
register_namespace('netconf', BASE_NS)
39

    
40
qualify = lambda tag, ns=BASE_NS: tag if ns is None else '{%s}%s' % (ns, tag)
41

    
42
multiqualify = lambda tag, nslist=(BASE_NS, CISCO_BS): [qualify(tag, ns) for ns in nslist]
43

    
44
unqualify = lambda tag: tag[tag.rfind('}')+1:]
45

    
46
### XML with Python data structures
47

    
48
class DictTree:
49

    
50
    @staticmethod
51
    def Element(spec):
52
        if iselement(spec):
53
            return spec
54
        elif isinstance(spec, basestring):
55
            return XML.Element(spec)
56
        if not isinstance(spec, dict):
57
            raise ContentError("Invalid tree spec")
58
        if 'tag' in spec:
59
            ele = ET.Element(spec.get('tag'), spec.get('attributes', {}))
60
            ele.text = spec.get('text', '')
61
            ele.tail = spec.get('tail', '')
62
            subtree = spec.get('subtree', [])
63
            # might not be properly specified as list but may be dict
64
            if isinstance(subtree, dict):
65
                subtree = [subtree]
66
            for subele in subtree:
67
                ele.append(DictTree.Element(subele))
68
            return ele
69
        elif 'comment' in spec:
70
            return ET.Comment(spec.get('comment'))
71
        else:
72
            raise ContentError('Invalid tree spec')
73
    
74
    @staticmethod
75
    def XML(spec, encoding='utf-8'):
76
        Element.XML(DictTree.Element(spec), encoding)
77

    
78
class Element:
79
    
80
    @staticmethod
81
    def DictTree(ele):
82
        return {
83
            'tag': ele.tag,
84
            'attributes': ele.attrib,
85
            'text': ele.text,
86
            'tail': ele.tail,
87
            'subtree': [ Element.DictTree(child) for child in root.getchildren() ]
88
        }
89
    
90
    @staticmethod
91
    def XML(ele, encoding='utf-8'):
92
        xml = ET.tostring(ele, encoding)
93
        return xml if xml.startswith('<?xml') else '<?xml version="1.0" encoding="%s"?>\n%s' % (encoding, xml)
94

    
95
class XML:
96
    
97
    @staticmethod
98
    def DictTree(xml):
99
        return Element.DictTree(XML.Element(xml))
100
    
101
    @staticmethod
102
    def Element(xml):
103
        return ET.fromstring(xml)
104

    
105
dtree2ele = DictTree.Element
106
dtree2xml = DictTree.XML
107
ele2dtree = Element.DictTree
108
ele2xml = Element.XML
109
xml2dtree = XML.DictTree
110
xml2ele = XML.Element
111

    
112
### Other utility functions
113

    
114
iselement = ET.iselement
115

    
116
def find(ele, tag, strict=False):
117
    """In strict mode, doesn't workaround Cisco implementations sending incorrectly
118
    namespaced XML. Supply qualified tag name if using strict mode.
119
    """
120
    if strict:
121
        return ele.find(tag)
122
    else:
123
        for qname in multiqualify(tag):
124
            found = ele.find(qname)
125
            if found is not None:
126
                return found
127

    
128
def parse_root(raw):
129
    '''Parse the top-level element from XML string.
130
    
131
    Returns a `(tag, attributes)` tuple, where `tag` is a string representing
132
    the qualified name of the root element and `attributes` is an
133
    `{attribute: value}` dictionary.
134
    '''
135
    fp = StringIO(raw[:1024]) # this is a guess but start element beyond 1024 bytes would be a bit absurd
136
    for event, element in ET.iterparse(fp, events=('start',)):
137
        return (element.tag, element.attrib)
138

    
139
def validated_element(rep, tag, attrs=None):
140
    ele = dtree2ele(rep)
141
    if ele.tag not in (tag, qualify(tag)):
142
        raise ContentError("Required root element [%s] not found" % tag)
143
    if attrs is not None:
144
        for req in attrs:
145
            for attr in ele.attrib:
146
                if unqualify(attr) == req:
147
                    break
148
            else:
149
                raise ContentError("Required attribute [%s] not found in element [%s]" % (req, req_tag))
150
    return ele