Implement collections.abc.Mapping for Multiaddr to give it a dict-style feel

Nit: All operations on this “dictionary” (including __len__ and __getitem__) are O(n) over the number of
bytes/protocols of the MultiAddr, rather then using more optimized code that would require pre-parsing
the binary MultiAddr. Considering that MultiAddr tend to be rather short this should hopefully not be
an issue, but changing this would be possible in the future without changing any part of the interface.

Author: ntninja

multiaddr/multiaddr.py

@@ -1,4 +1,9 @@
 # -*- coding: utf-8 -*-
+try:
+	import collections.abc
+except ImportError:  # pragma: no cover (PY2)
+	import collections
+	collections.abc = collections
 from copy import copy
 
 import six
@@ -10,7 +15,76 @@
 from .transforms import bytes_to_string
 
 
-class Multiaddr(object):
+__all__ = ("Multiaddr",)
+
+
+
+class MultiAddrKeys(collections.abc.KeysView, collections.abc.Sequence):
+    def __contains__(self, proto):
+        proto = protocols.protocol_with_any(proto)
+        return collections.abc.Sequence.__contains__(self, proto)
+
+    def __getitem__(self, idx):
+        if idx < 0:
+            idx = len(self)+idx
+        for idx2, proto in enumerate(self):
+            if idx2 == idx:
+                return proto
+        raise IndexError("Protocol list index out of range")
+
+    __hash__ = collections.abc.KeysView._hash
+
+    def __iter__(self):
+        for proto, _, _ in bytes_iter(self._mapping.to_bytes()):
+            yield proto
+
+
+class MultiAddrItems(collections.abc.ItemsView, collections.abc.Sequence):
+    def __contains__(self, item):
+        proto, value = item
+        proto = protocols.protocol_with_any(proto)
+        return collections.abc.Sequence.__contains__(self, (proto, value))
+
+    def __getitem__(self, idx):
+        if idx < 0:
+            idx = len(self)+idx
+        for idx2, item in enumerate(self):
+            if idx2 == idx:
+                return item
+        raise IndexError("Protocol item list index out of range")
+
+    def __iter__(self):
+        for proto, codec, part in bytes_iter(self._mapping.to_bytes()):
+            if codec.SIZE != 0:
+                try:
+                    # If we have an address, return it
+                    yield proto, codec.to_string(proto, part)
+                except Exception as exc:
+                    six.raise_from(exceptions.BinaryParseError(str(exc), self._mapping.to_bytes(), proto.name, exc), exc)
+            else:
+                # We were given something like '/utp', which doesn't have
+                # an address, so return None
+                yield proto, None
+
+
+class MultiAddrValues(collections.abc.ValuesView, collections.abc.Sequence):
+    __contains__ = collections.abc.Sequence.__contains__
+
+    def __getitem__(self, idx):
+        if idx < 0:
+            idx = len(self)+idx
+        for idx2, proto in enumerate(self):
+            if idx2 == idx:
+                return proto
+        raise IndexError("Protocol value list index out of range")
+
+    def __iter__(self):
+        for _, value in MultiAddrItems(self._mapping):
+            yield value
+
+
+
+class Multiaddr(collections.abc.Mapping):
     """Multiaddr is a representation of multiple nested internet addresses.
 
     Multiaddr is a cross-protocol, cross-platform format for representing
@@ -53,16 +127,22 @@ def __eq__(self, other):
         """Checks if two Multiaddr objects are exactly equal."""
         return self._bytes == other._bytes
 
-    def __ne__(self, other):
-        return not (self == other)
-
     def __str__(self):
         """Return the string representation of this Multiaddr.
 
         May raise a :class:`~multiaddr.exceptions.BinaryParseError` if the
         stored MultiAddr binary representation is invalid."""
         return bytes_to_string(self._bytes)
 
+    def __contains__(self, proto):
+        return proto in MultiAddrKeys(self)
+
+    def __iter__(self):
+        return iter(MultiAddrKeys(self))
+
+    def __len__(self):
+        return sum((1 for _ in bytes_iter(self.to_bytes())))
+
     # On Python 2 __str__ needs to return binary text, so expose the original
     # function as __unicode__ and transparently encode its returned text based
     # on the current locale
@@ -82,7 +162,15 @@ def to_bytes(self):
 
     def protocols(self):
         """Returns a list of Protocols this Multiaddr includes."""
-        return list(proto for proto, _, _ in bytes_iter(self.to_bytes()))
+        return MultiAddrKeys(self)
+
+    keys = protocols
+
+    def items(self):
+        return MultiAddrItems(self)
+
+    def values(self):
+        return MultiAddrValues(self)
 
     def encapsulate(self, other):
         """Wrap this Multiaddr around another.
@@ -125,24 +213,10 @@ def value_for_protocol(self, proto):
         ~multiaddr.exceptions.ProtocolLookupError
             MultiAddr does not contain any instance of this protocol
         """
-        if not isinstance(proto, protocols.Protocol):
-            if isinstance(proto, int):
-                proto = protocols.protocol_with_code(proto)
-            elif isinstance(proto, six.string_types):
-                proto = protocols.protocol_with_name(proto)
-            else:
-                raise TypeError("Protocol object, name or code expected, got {0!r}".format(proto))
-
-        for proto2, codec, part in bytes_iter(self.to_bytes()):
-            if proto2 == proto:
-                if codec.SIZE != 0:
-                    try:
-                        # If we have an address, return it
-                        return codec.to_string(proto2, part)
-                    except Exception as exc:
-                        six.raise_from(exceptions.BinaryParseError(str(exc), self.to_bytes(), proto2.name, exc), exc)
-                else:
-                    # We were given something like '/utp', which doesn't have
-                    # an address, so return None
-                    return None
+        proto = protocols.protocol_with_any(proto)
+        for proto2, value in self.items():
+            if proto2 is proto or proto2 == proto:
+                return value
         raise exceptions.ProtocolLookupError(proto, str(self))
+
+    __getitem__ = value_for_protocol

multiaddr/protocols.py

@@ -100,13 +100,16 @@ def vcode(self):
         return varint.encode(self.code)
 
     def __eq__(self, other):
+        if not isinstance(other, Protocol):
+            return NotImplemented
+
         return all((self.code == other.code,
                     self.name == other.name,
                     self.codec == other.codec,
                     self.path == other.path))
 
-    def __ne__(self, other):
-        return not self == other
+    def __hash__(self):
+        return self.code
 
     def __repr__(self):
         return "Protocol(code={code!r}, name={name!r}, codec={codec!r})".format(
@@ -200,6 +203,17 @@ def protocol_with_code(code):
     return _codes_to_protocols[code]
 
 
+def protocol_with_any(proto):
+    if isinstance(proto, Protocol):
+        return proto
+    elif isinstance(proto, int):
+        return protocol_with_code(proto)
+    elif isinstance(proto, six.string_types):
+        return protocol_with_name(proto)
+    else:
+        raise TypeError("Protocol object, name or code expected, got {0!r}".format(proto))
+
+
 def protocols_with_string(string):
     """Return a list of protocols matching given string."""
     # Normalize string

tests/test_multiaddr.py

@@ -234,6 +234,34 @@ def test_get_value():
     assert_value_for_proto(a, P_UNIX, "/studio")  # only a path.
 
 
+def test_views():
+    ma = Multiaddr(
+        "/ip4/127.0.0.1/utp/tcp/5555/udp/1234/utp/"
+        "p2p/QmbHVEEepCi7rn7VL7Exxpd2Ci9NNB6ifvqwhsrbRMgQFP")
+
+    for idx, (proto1, proto2, item, value) in enumerate(zip(ma, ma.keys(), ma.items(), ma.values())):
+        assert (proto1, value) == (proto2, value) == item
+        assert proto1 in ma
+        assert proto2 in ma.keys()
+        assert item in ma.items()
+        assert value in ma.values()
+        assert ma.keys()[idx] == ma.keys()[idx-len(ma)] == proto1 == proto2
+        assert ma.items()[idx] == ma.items()[idx-len(ma)] == item
+        assert ma.values()[idx] == ma.values()[idx-len(ma)] == ma[proto1] == value
+
+    assert len(ma.keys()) == len(ma.items()) == len(ma.values()) == len(ma)
+    assert len(list(ma.keys())) == len(ma.keys())
+    assert len(list(ma.items())) == len(ma.items())
+    assert len(list(ma.values())) == len(ma.values())
+    
+    with pytest.raises(IndexError):
+        ma.keys()[len(ma)]
+    with pytest.raises(IndexError):
+        ma.items()[len(ma)]
+    with pytest.raises(IndexError):
+        ma.values()[len(ma)]
+
+
 def test_bad_initialization_no_params():
     with pytest.raises(TypeError):
         Multiaddr()

tests/test_protocols.py

@@ -72,6 +72,7 @@ def test_protocol_with_name():
     assert proto.code == protocols.P_IP4
     assert proto.size == 32
     assert proto.vcode == varint.encode(protocols.P_IP4)
+    assert hash(proto) == protocols.P_IP4
 
     with pytest.raises(exceptions.ProtocolNotFoundError):
         proto = protocols.protocol_with_name('foo')
@@ -83,6 +84,7 @@ def test_protocol_with_code():
     assert proto.code == protocols.P_IP4
     assert proto.size == 32
     assert proto.vcode == varint.encode(protocols.P_IP4)
+    assert hash(proto) == protocols.P_IP4
 
     with pytest.raises(exceptions.ProtocolNotFoundError):
         proto = protocols.protocol_with_code(1234)
@@ -95,6 +97,8 @@ def test_protocol_equality():
 
     assert proto1 == proto2
     assert proto1 != proto3
+    assert proto1 != None
+    assert proto2 != str(proto2)
 
 
 @pytest.mark.parametrize("names", [['ip4'],