Merge pull request #163 from aldanor/feature/from-bytes-buffer

Support generic buffers in from_bytes()

capnp/lib/capnp.pyx

@@ -15,7 +15,7 @@ from libc.stdlib cimport malloc, free
 from libc.string cimport memcpy
 from cython.operator cimport dereference as deref
 from cpython.exc cimport PyErr_Clear
-from cpython cimport Py_buffer
+from cpython cimport Py_buffer, PyObject_CheckBuffer
 from cpython.buffer cimport PyBUF_SIMPLE
 
 from types import ModuleType as _ModuleType
@@ -31,7 +31,6 @@ import threading as _threading
 import socket as _socket
 import random as _random
 import collections as _collections
-import mmap as _mmap
 
 _CAPNP_VERSION_MAJOR = capnp.CAPNP_VERSION_MAJOR
 _CAPNP_VERSION_MINOR = capnp.CAPNP_VERSION_MINOR
@@ -3756,11 +3755,7 @@ cdef class _FlatArrayMessageReader(_MessageReader):
             raise ValueError("input length must be a multiple of eight bytes")
 
         cdef char * ptr
-        if type(buf) == _mmap.mmap:
-            view = _BufferView(buf)
-            ptr = view.buf
-            self._object_to_pin = view
-        else:
+        if isinstance(buf, bytes):
             ptr = buf
             if (<uintptr_t>ptr) % 8 != 0:
                 aligned = _AlignedBuffer(buf)
@@ -3768,6 +3763,12 @@ cdef class _FlatArrayMessageReader(_MessageReader):
                 self._object_to_pin = aligned
             else:
                 self._object_to_pin = buf
+        elif PyObject_CheckBuffer(buf):
+            view = _BufferView(buf)
+            ptr = view.buf
+            self._object_to_pin = view
+        else:
+            raise TypeError('expected buffer-like object in FlatArrayMessageReader')
 
         self.thisptr = new schema_cpp.FlatArrayMessageReader(
             schema_cpp.WordArrayPtr(<schema_cpp.word*>ptr, sz//8),

test/test_serialization.py

@@ -65,6 +65,20 @@ def test_roundtrip_bytes_mmap(all_types):
         msg = all_types.TestAllTypes.from_bytes(memory)
         test_regression.check_all_types(msg)
 
+@pytest.mark.skipif(sys.version_info[0] < 3, reason="memoryview is a builtin on Python 3")
+def test_roundtrip_bytes_buffer(all_types):
+    msg = all_types.TestAllTypes.new_message()
+    test_regression.init_all_types(msg)
+
+    b = msg.to_bytes()
+    v = memoryview(b)
+    msg = all_types.TestAllTypes.from_bytes(v)
+    test_regression.check_all_types(msg)
+
+def test_roundtrip_bytes_fail(all_types):
+    with pytest.raises(TypeError):
+        all_types.TestAllTypes.from_bytes(42)
+
 @pytest.mark.skipif(platform.python_implementation() == 'PyPy', reason="This works in PyPy 4.0.1 but travisci's version of PyPy has some bug that fails this test.")
 def test_roundtrip_bytes_packed(all_types):
     msg = all_types.TestAllTypes.new_message()