openstack/deb-python-colander
Code Issues Proposed changes

Implement flatten.

Browse Source
This commit is contained in:
Chris Rossi
2011-07-21 15:51:03 -04:00
parent db9dc196f0
commit 3ec2a0933e
3 changed files with 325 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
CHANGES.txt
View File

@@ -1,6 +1,16 @@
Changes
=======
Unreleased
----------
- ``flatten`` now only includes leaf nodes in the flattened dict.
- ``flatten`` does not include a path element for the name of the type node
for sequences.
- ``unflatten`` is implemented.
0.9.3 (2011-06-23)
------------------

112
colander/__init__.py
View File

@@ -336,12 +336,20 @@ class OneOf(object):
class SchemaType(object):
""" Base class for all schema types """
def flatten(self, node, appstruct, prefix=''):
def flatten(self, node, appstruct, prefix='', listitem=False):
result = {}
if listitem:
selfname = prefix
else:
selfname = '%s%s' % (prefix, node.name)
result[selfname] = appstruct
return result
def unflatten(self, node, paths, fstruct):
name = node.name
assert paths == [name], "paths should be [name] for leaf nodes."
return fstruct[name]
class Mapping(SchemaType):
""" A type which represents a mapping of names to nodes.
@@ -462,11 +470,12 @@ class Mapping(SchemaType):
return self._impl(node, cstruct, callback)
def flatten(self, node, appstruct, prefix=''):
def flatten(self, node, appstruct, prefix='', listitem=False):
result = {}
selfname = '%s%s' % (prefix, node.name)
selfprefix = selfname + '.'
result[selfname] = appstruct
if listitem:
selfprefix = prefix
else:
selfprefix = '%s%s.' % (prefix, node.name)
for subnode in node.children:
name = subnode.name
@@ -475,6 +484,10 @@ class Mapping(SchemaType):
prefix=selfprefix))
return result
def unflatten(self, node, paths, fstruct):
return _unflatten_mapping(node, paths, fstruct)
class Positional(object):
"""
Marker abstract base class meaning 'this type has children which
@@ -554,11 +567,12 @@ class Tuple(Positional, SchemaType):
return self._impl(node, cstruct, callback)
def flatten(self, node, appstruct, prefix=''):
def flatten(self, node, appstruct, prefix='', listitem=False):
result = {}
selfname = '%s%s' % (prefix, node.name)
selfprefix = selfname + '.'
result[selfname] = appstruct
if listitem:
selfprefix = prefix
else:
selfprefix = '%s%s.' % (prefix, node.name)
for num, subnode in enumerate(node.children):
substruct = appstruct[num]
@@ -566,6 +580,13 @@ class Tuple(Positional, SchemaType):
prefix=selfprefix))
return result
def unflatten(self, node, paths, fstruct):
mapstruct = _unflatten_mapping(node, paths, fstruct)
appstruct = []
for subnode in node.children:
appstruct.append(mapstruct[subnode.name])
return tuple(appstruct)
class Sequence(Positional, SchemaType):
"""
A type which represents a variable-length sequence of nodes,
@@ -685,23 +706,38 @@ class Sequence(Positional, SchemaType):
return self._impl(node, cstruct, callback, accept_scalar)
def flatten(self, node, appstruct, prefix=''):
def flatten(self, node, appstruct, prefix='', listitem=False):
result = {}
selfname = '%s%s' % (prefix, node.name)
selfprefix = selfname + '.'
result[selfname] = appstruct
if listitem:
selfprefix = prefix
else:
selfprefix = '%s%s.' % (prefix, node.name)
childnode = node.children[0]
for num, subval in enumerate(appstruct):
subname = '%s%s' % (selfprefix, num)
result[subname] = subval
subprefix = subname + '.'
result.update(childnode.typ.flatten(childnode, subval,
prefix=subprefix))
result.update(childnode.typ.flatten(
childnode, subval, prefix=subprefix, listitem=True))
return result
def unflatten(self, node, paths, fstruct):
only_child = node.children[0]
child_name = only_child.name
def get_child(name):
return only_child
def rewrite_subpath(subpath):
if '.' in subpath:
suffix = subpath.split('.', 1)[1]
return '%s.%s' % (child_name, suffix)
return child_name
mapstruct = _unflatten_mapping(node, paths, fstruct,
get_child, rewrite_subpath)
return [mapstruct[str(index)] for index in xrange(len(mapstruct))]
Seq = Sequence
class String(SchemaType):
@@ -1422,6 +1458,8 @@ class SchemaNode(object):
def unflatten(self, fstruct):
""" Create an appstruct based on the schema represented by
this node using the fstruct passed. """
paths = sorted(fstruct.keys())
return self.typ.unflatten(self, paths, fstruct)
def deserialize(self, cstruct=null):
""" Deserialize the :term:`cstruct` into an :term:`appstruct` based
@@ -1599,3 +1637,45 @@ class deferred(object):
def __call__(self, node, kw):
return self.wrapped(node, kw)
def _unflatten_mapping(node, paths, fstruct,
get_child=None, rewrite_subpath=None):
if get_child is None:
get_child = node.__getitem__
if rewrite_subpath is None:
def rewrite_subpath(subpath):
return subpath
node_name = node.name
prefix = node_name + '.'
prefix_len = len(prefix)
appstruct = {}
subfstruct = {}
subpaths = []
curname = None
for path in paths:
if path == node_name:
# flattened structs contain non-leaf nodes which are ignored
# during unflattening.
continue
assert path.startswith(prefix), "Bad node: %s" % path
subpath = path[prefix_len:]
if '.' in subpath:
name = subpath[:subpath.index('.')]
else:
name = subpath
if curname is None:
curname = name
elif name != curname:
subnode = get_child(curname)
appstruct[curname] = subnode.typ.unflatten(
subnode, subpaths, subfstruct)
subfstruct = {}
subpaths = []
curname = name
subpath = rewrite_subpath(subpath)
subfstruct[subpath] = fstruct[path]
subpaths.append(subpath)
if curname is not None:
subnode = get_child(curname)
appstruct[curname] = subnode.typ.unflatten(
subnode, subpaths, subfstruct)
return appstruct

249
colander/tests.py
View File

@@ -330,6 +330,17 @@ class TestSchemaType(unittest.TestCase):
result = typ.flatten(node, 'appstruct')
self.assertEqual(result, {'node':'appstruct'})
def test_flatten_listitem(self):
node = DummySchemaNode(None, name='node')
typ = self._makeOne()
result = typ.flatten(node, 'appstruct', listitem=True)
self.assertEqual(result, {'':'appstruct'})
def test_unflatten(self):
node = DummySchemaNode(None, name='node')
typ = self._makeOne()
result = typ.unflatten(node, ['node'], {'node': 'appstruct'})
self.assertEqual(result, 'appstruct')
class TestMapping(unittest.TestCase):
def _makeOne(self, *arg, **kw):
@@ -467,8 +478,62 @@ class TestMapping(unittest.TestCase):
]
typ = self._makeOne()
result = typ.flatten(node, {'a':1, 'b':2})
self.assertEqual(result,
{'node': {'a': 1, 'b': 2}, 'node.appstruct': 2})
self.assertEqual(result, {'node.appstruct': 2})
def test_flatten_listitem(self):
node = DummySchemaNode(None, name='node')
int1 = DummyType()
int2 = DummyType()
node.children = [
DummySchemaNode(int1, name='a'),
DummySchemaNode(int2, name='b'),
]
typ = self._makeOne()
result = typ.flatten(node, {'a':1, 'b':2}, listitem=True)
self.assertEqual(result, {'appstruct': 2})
def test_unflatten(self):
node = DummySchemaNode(None, name='node')
int1 = DummyType()
int2 = DummyType()
node.children = [
DummySchemaNode(int1, name='a'),
DummySchemaNode(int2, name='b'),
]
typ = self._makeOne()
result = typ.unflatten(node,
['node', 'node.a', 'node.b'],
{'node': {'a':1, 'b':2}, 'node.a':1, 'node.b':2})
self.assertEqual(result, {'a': 1, 'b': 2})
def test_unflatten_nested(self):
node = DummySchemaNode(None, name='node')
inttype = DummyType()
one = DummySchemaNode(self._makeOne(), name='one')
one.children = [
DummySchemaNode(inttype, name='a'),
DummySchemaNode(inttype, name='b'),
]
two = DummySchemaNode(self._makeOne(), name='two')
two.children = [
DummySchemaNode(inttype, name='c'),
DummySchemaNode(inttype, name='d'),
]
node.children = [one, two]
typ = self._makeOne()
result = typ.unflatten(
node, ['node', 'node.one', 'node.one.a', 'node.one.b',
'node.two', 'node.two.c', 'node.two.d'],
{'node': {'one': {'a': 1, 'b': 2}, 'two': {'c': 3, 'd': 4}},
'node.one': {'a': 1, 'b': 2},
'node.two': {'c': 3, 'd': 4},
'node.one.a': 1,
'node.one.b': 2,
'node.two.c': 3,
'node.two.d': 4,})
self.assertEqual(result, {
'one': {'a': 1, 'b': 2}, 'two': {'c': 3, 'd': 4}})
class TestTuple(unittest.TestCase):
def _makeOne(self):
@@ -598,7 +663,32 @@ class TestTuple(unittest.TestCase):
]
typ = self._makeOne()
result = typ.flatten(node, (1, 2))
self.assertEqual(result, {'node': (1, 2), 'node.appstruct': 2})
self.assertEqual(result, {'node.appstruct': 2})
def test_flatten_listitem(self):
node = DummySchemaNode(None, name='node')
int1 = DummyType()
int2 = DummyType()
node.children = [
DummySchemaNode(int1, name='a'),
DummySchemaNode(int2, name='b'),
]
typ = self._makeOne()
result = typ.flatten(node, (1, 2), listitem=True)
self.assertEqual(result, {'appstruct': 2})
def test_unflatten(self):
node = DummySchemaNode(None, name='node')
int1 = DummyType()
int2 = DummyType()
node.children = [
DummySchemaNode(int1, name='a'),
DummySchemaNode(int2, name='b'),
]
typ = self._makeOne()
result = typ.unflatten(node, ['node', 'node.a', 'node.b'],
{'node': (1, 2), 'node.a': 1, 'node.b': 2})
self.assertEqual(result, (1, 2))
class TestSequence(unittest.TestCase):
def _makeOne(self, **kw):
@@ -712,13 +802,31 @@ class TestSequence(unittest.TestCase):
]
typ = self._makeOne()
result = typ.flatten(node, [1, 2])
self.assertEqual(
result,
{'node': [1, 2],
'node.0': 1,
'node.0.appstruct': 1,
'node.1.appstruct': 2,
'node.1': 2})
self.assertEqual(result, {'node.0': 1, 'node.1': 2})
def test_flatten_listitem(self):
node = DummySchemaNode(None, name='node')
int1 = DummyType()
int2 = DummyType()
node.children = [
DummySchemaNode(int1, name='a'),
DummySchemaNode(int2, name='b'),
]
typ = self._makeOne()
result = typ.flatten(node, [1, 2], listitem=True)
self.assertEqual(result, {'0': 1, '1': 2})
def test_unflatten(self):
node = DummySchemaNode(None, name='node')
node.children = [
DummySchemaNode(DummyType(), name='foo'),
]
typ = self._makeOne()
result = typ.unflatten(node,
['node.0', 'node.1',],
{'node.0': 'a', 'node.1': 'b'})
self.assertEqual(result, ['a', 'b'])
class TestString(unittest.TestCase):
def _makeOne(self, encoding=None):
@@ -1935,45 +2043,78 @@ class TestFunctional(object):
result = schema.flatten(appstruct)
expected = {
'schema.seq.2.tup.tupstring': 's',
'schema.seq2.0.mapping.key2': 2,
'schema.seq.0': (1, 's'),
'schema.seq.1': (2, 's'),
'schema.seq.2': (3, 's'),
'schema.seq.3': (4, 's'),
'schema.seq': [(1, 's'), (2, 's'), (3, 's'), (4, 's')],
'schema.seq.2.tupstring': 's',
'schema.seq2.0.key2': 2,
'schema.ob': colander.tests,
'schema.seq2.1.mapping.key2': 4,
'schema.seq.0.tup': (1, 's'),
'schema.seq.1.tup': (2, 's'),
'schema.seq2.0.mapping': {'key2': 2, 'key': 1},
'schema.seq2.1.mapping': {'key2': 4, 'key': 3},
'schema.seq.1.tup.tupstring': 's',
'schema.seq2.0.mapping.key': 1,
'schema.seq.1.tup.tupint': 2,
'schema.tup': (1, 's'),
'schema.seq.3.tup': (4, 's'),
'schema.seq.0.tup.tupstring': 's',
'schema.seq.2.tup': (3, 's'),
'schema.seq.3.tup.tupstring': 's',
'schema.seq.3.tup.tupint': 4,
'schema.seq2.1.mapping.key': 3,
'schema.seq2.1.key2': 4,
'schema.seq.1.tupstring': 's',
'schema.seq2.0.key': 1,
'schema.seq.1.tupint': 2,
'schema.seq.0.tupstring': 's',
'schema.seq.3.tupstring': 's',
'schema.seq.3.tupint': 4,
'schema.seq2.1.key': 3,
'schema.int': 10,
'schema.seq2.0': {'key2': 2, 'key': 1},
'schema.seq.0.tup.tupint': 1,
'schema.seq.0.tupint': 1,
'schema.tup.tupint': 1,
'schema.tup.tupstring': 's',
'schema.seq.2.tup.tupint': 3,
'schema.seq2': [{'key2': 2, 'key': 1}, {'key2': 4, 'key': 3}],
'schema.seq2.1': {'key2': 4, 'key': 3},
'schema': {'int': 10,
'seq2': [{'key2': 2, 'key': 1}, {'key2': 4, 'key': 3}],
'tup': (1, 's'),
'ob':colander.tests,
'seq': [(1, 's'), (2, 's'), (3, 's'), (4, 's')]}}
'schema.seq.2.tupint': 3,
}
for k, v in expected.items():
self.assertEqual(result[k], v)
for k, v in result.items():
self.assertEqual(v, expected[k])
self.assertEqual(expected[k], v)
def test_unflatten_ok(self):
import colander
fstruct = {
'schema.seq.2.tupstring': 's',
'schema.seq2.0.key2': 2,
'schema.ob': colander.tests,
'schema.seq2.1.key2': 4,
'schema.seq.1.tupstring': 's',
'schema.seq2.0.key': 1,
'schema.seq.1.tupint': 2,
'schema.seq.0.tupstring': 's',
'schema.seq.3.tupstring': 's',
'schema.seq.3.tupint': 4,
'schema.seq2.1.key': 3,
'schema.int': 10,
'schema.seq.0.tupint': 1,
'schema.tup.tupint': 1,
'schema.tup.tupstring': 's',
'schema.seq.2.tupint': 3,
}
schema = self._makeSchema()
result = schema.unflatten(fstruct)
expected = {
'int':10,
'ob':colander.tests,
'seq':[(1, 's'),(2, 's'), (3, 's'), (4, 's')],
'seq2':[{'key':1, 'key2':2}, {'key':3, 'key2':4}],
'tup':(1, 's'),
}
for k, v in expected.items():
self.assertEqual(result[k], v)
for k, v in result.items():
self.assertEqual(expected[k], v)
def test_flatten_unflatten_roundtrip(self):
import colander
appstruct = {
'int':10,
'ob':colander.tests,
'seq':[(1, 's'),(2, 's'), (3, 's'), (4, 's')],
'seq2':[{'key':1, 'key2':2}, {'key':3, 'key2':4}],
'tup':(1, 's'),
}
schema = self._makeSchema(name='')
self.assertEqual(
schema.unflatten(schema.flatten(appstruct)),
appstruct)
def test_invalid_asdict(self):
expected = {
@@ -2002,7 +2143,7 @@ class TestFunctional(object):
class TestImperative(unittest.TestCase, TestFunctional):
def _makeSchema(self):
def _makeSchema(self, name='schema'):
import colander
integer = colander.SchemaNode(
@@ -2059,14 +2200,13 @@ class TestImperative(unittest.TestCase, TestFunctional):
tup,
seq,
seq2,
name='schema')
name=name)
return schema
class TestDeclarative(unittest.TestCase, TestFunctional):
def _makeSchema(self):
def _makeSchema(self, name='schema'):
import colander
@@ -2092,7 +2232,7 @@ class TestDeclarative(unittest.TestCase, TestFunctional):
tup = TupleSchema()
seq2 = SequenceTwo()
schema = MainSchema(name='schema')
schema = MainSchema(name=name)
return schema
class Test_null(unittest.TestCase):
@@ -2133,6 +2273,11 @@ class DummySchemaNode(object):
raise Invalid(self, self.exc)
return val
def __getitem__(self, name):
for child in self.children:
if child.name == name:
return child
class DummyValidator(object):
def __init__(self, msg=None):
self.msg = msg
@@ -2155,7 +2300,13 @@ class DummyType(object):
def deserialize(self, node, value):
return value
def flatten(self, node, appstruct, prefix=''):
def flatten(self, node, appstruct, prefix='', listitem=False):
if listitem:
key = prefix.rstrip('.')
else:
key = prefix + 'appstruct'
return {key:appstruct}
def unflatten(self, node, paths, fstruct):
assert paths == [node.name]
return fstruct[node.name]