1. Killian De Smedt
  2. py3_protobuf

Commits

Killian De Smedt  committed fe2c451

Added package name parsing to use as namespace, use a namedtuple hierarchy to represent namespaces

  • Participants
  • Parent commits 1fe0dc9
  • Branches default

Comments (0)

Files changed (2)

File protobuf/lexer.py

View file
  • Ignore whitespace
             attrs[key] = int(next(tokens)[1])
     return name, type(name, (object,), attrs), tokens
 
+def _decode_namespace(tokens):
+    ns = []
+    while True:
+        key = next(tokens)[1]
+        if key == ';': break
+        else: ns.append(key)
+    return ns, None, tokens
+        
+
 _type_to_decoder = {'message': _decode_message,
                    'enum': _decode_enum,
                    'optional': _decode_attr,
                    'required': _decode_attr,
                    'repeated': partial(_decode_attr, repeated=True),
-                   '}': lambda x: (None, True, x)
+                   '}': lambda x: (None, True, x),
+                    'package': _decode_namespace
                    }
 
 def message_from_string(string, package_name='protobuf_package'):
     '''
     Creates a set of message classes from a string containing protobuf messages.
-    Note that import statements do not work and packages are ignored
+    Note that import statements do not work because there is no filesystem context known.
     '''
+    namespace = [package_name]
     lex(string, ProtoBufLexer())
     tokens = ((x, y) for x, y in lex(string, ProtoBufLexer()) if x not in [tok.Text, tok.Operator] + list(tok.Comment.subtypes))
     # currently do not support any top-level options or statements, skip to first declaration:
     attrs = {}
-    while True:
-        try:
-            tokens = dropwhile(lambda x: x[0] != tok.Keyword.Declaration, tokens)
+    try:
+        while True:
+            tokens = dropwhile(lambda x: x[0] not in [tok.Keyword.Declaration, tok.Keyword.Namespace], tokens)
             name, new_type, tokens = _type_to_decoder[next(tokens)[1]](tokens)
-            attrs[name] = new_type
-        except StopIteration:
-            break
-    return namedtuple(package_name, attrs.keys(), verbose=False)(**attrs)
+            if new_type:
+                attrs[name] = new_type
+            else:
+                namespace = name
+    except StopIteration:
+        pass
+    for ns_name in namespace[::-1]:
+        result = namedtuple(ns_name, attrs.keys(), verbose=False)(**attrs)
+        attrs = {ns_name: result}
+    return result

File protobuf_test/test_lexer.py

View file
  • Ignore whitespace
         self.assertEqual(Test(test=150).serialize_to_string(), b'\x08\x96\x01')
 
 
+    def test_package(self):
+        msg = """
+        package Foo;
+        message Test {
+            optional int32 test = 1;
+        }
+        """
+        Bar = message_from_string(msg)
+        self.assertEqual(Bar.__class__.__name__,'Foo')
+        self.assertEqual(Bar.Test(test=150).serialize_to_string(), b'\x08\x96\x01')
+        msg = """
+        package Foo.Bar;
+        message Test {
+            optional int32 test = 1;
+        }
+        """
+        Foo = message_from_string(msg)
+        self.assertEqual(Foo.__class__.__name__,'Foo')
+        self.assertEqual(Foo.Bar.__class__.__name__,'Bar')
+        self.assertEqual(Foo.Bar.Test(test=150).serialize_to_string(), b'\x08\x96\x01')
+
 if __name__ == "__main__":
     # import sys;sys.argv = ['', 'Test.testName']
     unittest.main()