feat: add support for Python 3.5

Add support for Python 3.5 via async_generator library.

.travis.yml

@@ -9,15 +9,22 @@ stages:
   - name: deploy
     if: tag IS present AND type != pull_request
 env:
+  # We run pypy tests first because they run the slowest
+  - PYTHON_IMAGE="pypy:3 PURERPC_BACKEND="asyncio"
+  - PYTHON_IMAGE="pypy:3" PURERPC_BACKEND="curio"
+  - PYTHON_IMAGE="pypy:3" PURERPC_BACKEND="trio"
+  - PYTHON_IMAGE="standy/pypy:3.6-6.1.0" PURERPC_BACKEND="asyncio"
+  - PYTHON_IMAGE="standy/pypy:3.6-6.1.0" PURERPC_BACKEND="curio"
+  - PYTHON_IMAGE="standy/pypy:3.6-6.1.0" PURERPC_BACKEND="trio"
+  - PYTHON_IMAGE="python:3.5" PURERPC_BACKEND="asyncio"
+  - PYTHON_IMAGE="python:3.5" PURERPC_BACKEND="curio"
+  - PYTHON_IMAGE="python:3.5" PURERPC_BACKEND="trio"
   - PYTHON_IMAGE="python:3.6" PURERPC_BACKEND="asyncio"
   - PYTHON_IMAGE="python:3.6" PURERPC_BACKEND="curio"
   - PYTHON_IMAGE="python:3.6" PURERPC_BACKEND="trio"
   - PYTHON_IMAGE="python:3.7" PURERPC_BACKEND="asyncio"
   - PYTHON_IMAGE="python:3.7" PURERPC_BACKEND="curio"
   - PYTHON_IMAGE="python:3.7" PURERPC_BACKEND="trio"
-  - PYTHON_IMAGE="standy/pypy:3.6-6.1.0" PURERPC_BACKEND="asyncio"
-  - PYTHON_IMAGE="standy/pypy:3.6-6.1.0" PURERPC_BACKEND="curio"
-  - PYTHON_IMAGE="standy/pypy:3.6-6.1.0" PURERPC_BACKEND="trio"
 
 script:
   - ./ci/run_tests_in_docker.sh $PYTHON_IMAGE $PURERPC_BACKEND

README.md

@@ -9,8 +9,8 @@ Asynchronous pure Python gRPC client and server implementation supporting
 
 ## Requirements
 
-* CPython >= 3.6
-* PyPy >= 3.6
+* CPython >= 3.5
+* PyPy >= 3.5
 
 ## Installation
 
@@ -45,7 +45,11 @@ python -m grpc_tools.protoc --purerpc_out=. --python_out=. -I. greeter.proto
 
 ## Usage
 
-NOTE: `greeter_grpc` module is generated by purerpc's `protoc-gen-purerpc` plugin
+NOTE: `greeter_grpc` module is generated by purerpc's `protoc-gen-purerpc` plugin.
+
+Below is the examples for Python 3.6 and above which introduced asynchronous generators as a language concept.
+For Python 3.5, where native asynchronous generators are not supported, you can use [async_generator](https://github.com/python-trio/async_generator) library for this purpose.
+Just mark yielding coroutines with `@async_generator` decorator and use `await yield_(value)` and `await yield_from_(async_iterable)` instead of `yield`.
 
 ### Server
 

ci/run_tests_in_docker.sh

@@ -8,5 +8,12 @@ BUILD_TAG=${BASE_IMAGE//:/-}
 BUILD_TAG=${BUILD_TAG//\//-}
 
 docker build --build-arg BASE_IMAGE=${BASE_IMAGE} -t "standy/purerpc:${BUILD_TAG}" .
+
 echo "Runnig tests with $PURERPC_BACKEND backend"
-docker run -it -e PURERPC_BACKEND=${PURERPC_BACKEND} "standy/purerpc:$BUILD_TAG" bash -c 'python setup.py test'
+if [[ $BASE_IMAGE == pypy* ]]; then
+  CMD="pypy3 setup.py test"
+else
+  CMD="python setup.py test"
+fi
+
+docker run -it -e PURERPC_BACKEND=${PURERPC_BACKEND} "standy/purerpc:$BUILD_TAG" bash -c "$CMD"

setup.py

@@ -46,13 +46,14 @@ def main():
         packages=find_packages('src'),
         package_dir={'': 'src'},
         test_suite="tests",
-        python_requires=">=3.6.0",
+        python_requires=">=3.5",
         install_requires=[
-            "h2",
-            "protobuf",
-            "anyio",
-            "async_exit_stack",
-            "tblib",
+            "h2>=3.1.0,<4",
+            "protobuf~=3.6",
+            "anyio>=1.0.0b1,<2",
+            "async_exit_stack>=1.0.1,<2",
+            "tblib>=1.3.2,<2",
+            "async_generator>=1.10,<2.0"
         ],
         entry_points={'console_scripts': console_scripts},
         setup_requires=["pytest-runner"],

src/purerpc/anyio_monkeypatch.py

@@ -14,7 +14,7 @@
 def _new_run(func, *args, backend=None, backend_options=None):
     if backend is None:
         backend = os.getenv("PURERPC_BACKEND", "asyncio")
-    log.info(f"Selected {backend} backend")
+    log.info("Selected {} backend".format(backend))
     _anyio_run(func, *args, backend=backend, backend_options=backend_options)
 
 

src/purerpc/grpc_socket.py

@@ -3,6 +3,7 @@
 import datetime
 
 import anyio
+from async_generator import async_generator, yield_, yield_from_
 import h2
 import h2.events
 import h2.exceptions
@@ -193,6 +194,7 @@ def _decref_stream(self, stream_id: int):
             del self._streams[stream_id]
             del self._streams_count[stream_id]
 
+    @async_generator
     async def _listen(self):
         while True:
             data = await self._socket.recv(self._receive_buffer_size)
@@ -214,7 +216,7 @@ async def _listen(self):
                 await self._streams[event.stream_id]._incoming_events.put(event)
 
                 if isinstance(event, RequestReceived):
-                    yield self._streams[event.stream_id]
+                    await yield_(self._streams[event.stream_id])
                 elif isinstance(event, ResponseEnded) or isinstance(event, RequestEnded):
                     self._decref_stream(event.stream_id)
             await self._socket.try_flush()
@@ -229,11 +231,11 @@ async def initiate_connection(self, parent_task_group: anyio.abc.TaskGroup):
         if self.client_side:
             await parent_task_group.spawn(self._listener_thread)
 
+    @async_generator
     async def listen(self):
         if self.client_side:
             raise ValueError("Cannot listen client-side socket")
-        async for stream in self._listen():
-            yield stream
+        await yield_from_(self._listen())
 
     async def start_request(self, scheme: str, service_name: str, method_name: str,
                             message_type=None, authority=None, timeout: datetime.timedelta=None,

src/purerpc/grpclib/buffers.py

@@ -100,8 +100,12 @@ def _parse_one_message(self):
             if self._message_length > self._max_message_length:
                 # Even after the error is raised, the state is not corrupted, and parsing
                 # can be safely resumed
-                raise MessageTooLargeError("Received message larger than max: "
-                                           f"{self._message_length} > {self._max_message_length}")
+                raise MessageTooLargeError(
+                    "Received message larger than max: {message_length} > {max_message_length}".format(
+                        message_length=self._message_length,
+                        max_message_length=self._max_message_length,
+                    )
+                )
         if len(self._buffer) < self._message_length:
             return None, 0
         data, flow_controlled_length = self._buffer.popleft_flowcontrol(self._message_length)
@@ -160,8 +164,12 @@ def write_message(self, data: bytes, compress=False):
         if compress:
             data = self.compress(data)
         if len(data) > self._max_message_length:
-            raise MessageTooLargeError("Trying to send message larger than max: "
-                                       f"{len(data)} > {self._max_message_length}")
+            raise MessageTooLargeError(
+                "Trying to send message larger than max: {message_length} > {max_message_length}".format(
+                    message_length=len(data),
+                    max_message_length=self._max_message_length,
+                )
+            )
         self._buffer.append(struct.pack('>?I', compress, len(data)))
         self._buffer.append(data)
 

src/purerpc/grpclib/events.py

@@ -89,8 +89,13 @@ def parse_from_stream_id_and_headers_destructive(stream_id: int, headers: Header
         return event
 
     def __repr__(self):
-        return (f"<purerpc.grpclib.events.RequestReceived stream_id: {self.stream_id}, "\
-                f"service_name: {self.service_name}, method_name: {self.method_name}>")
+        fmt_string = ("<purerpc.grpclib.events.RequestReceived stream_id: {stream_id}, "
+                      "service_name: {service_name}, method_name: {method_name}>")
+        return fmt_string.format(
+            stream_id=self.stream_id,
+            service_name=self.service_name,
+            method_name=self.method_name,
+        )
 
 
 class MessageReceived(Event):
@@ -100,16 +105,23 @@ def __init__(self, stream_id: int, data: bytes, flow_controlled_length: int):
         self.flow_controlled_length = flow_controlled_length
 
     def __repr__(self):
-        return (f"<purerpc.grpclib.events.MessageReceived stream_id: {self.stream_id}, "
-                f"flow_controlled_length: {self.flow_controlled_length}>")
+        fmt_string= ("<purerpc.grpclib.events.MessageReceived stream_id: {stream_id}, "
+                     "flow_controlled_length: {flow_controlled_length}>")
+        return fmt_string.format(
+            stream_id=self.stream_id,
+            flow_controlled_length=self.flow_controlled_length,
+        )
 
 
 class RequestEnded(Event):
     def __init__(self, stream_id: int):
         self.stream_id = stream_id
 
     def __repr__(self):
-        return f"<purerpc.grpclib.events.RequestEnded stream_id: {self.stream_id}>"
+        fmt_string = "<purerpc.grpclib.events.RequestEnded stream_id: {stream_id}>"
+        return fmt_string.format(
+            stream_id=self.stream_id,
+        )
 
 
 class ResponseReceived(Event):
@@ -142,7 +154,11 @@ def parse_from_stream_id_and_headers_destructive(stream_id: int, headers: Header
         return event
 
     def __repr__(self):
-        return f"<purerpc.grpclib.events.ResponseReceived stream_id: {self.stream_id} content_type: {self.content_type}>"
+        fmt_string = "<purerpc.grpclib.events.ResponseReceived stream_id: {stream_id} content_type: {content_type}>"
+        return fmt_string.format(
+            stream_id=self.stream_id,
+            content_type=self.content_type,
+        )
 
 
 class ResponseEnded(Event):
@@ -170,4 +186,8 @@ def parse_from_stream_id_and_headers_destructive(stream_id: int, headers: Header
         return event
 
     def __repr__(self):
-        return f"<purerpc.grpclib.events.ResponseEnded stream_id: {self.stream_id}, status: {self.status}>"
+        fmt_string = "<purerpc.grpclib.events.ResponseEnded stream_id: {stream_id}, status: {status}>"
+        return fmt_string.format(
+            stream_id=self.stream_id,
+            status=self.status,
+        )