Merge pull request #14 from standy66/py3.5

feat: add support for Python 3.5

Author: standy66

.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

@@ -1,6 +1,7 @@
 # purerpc
 
 [![Build Status](https://travis-ci.org/standy66/purerpc.png?branch=master)](https://travis-ci.org/standy66/purerpc)
+[![Downloads](https://pepy.tech/badge/purerpc/month)](https://pepy.tech/project/purerpc)
 
 Asynchronous pure Python gRPC client and server implementation supporting
 [asyncio](https://docs.python.org/3/library/asyncio.html),
@@ -9,8 +10,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 +46,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 are 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
 
@@ -94,7 +99,7 @@ async def main():
 
 
 if __name__ == "__main__":
-    curio.run(main)  # Or trio.run(main)
+    curio.run(main)  # Or trio.run(main), or run in asyncio event loop
 ```
 
 You can mix server and client code, for example make a server that requests something using purerpc from another gRPC server, etc.

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

@@ -2,6 +2,8 @@
 import os
 from setuptools import setup, find_packages
 
+exec(open("src/purerpc/_version.py", encoding="utf-8").read())
+
 
 def read(*names, **kwargs):
     with open(os.path.join(os.path.dirname(__file__), *names), "r") as fin:
@@ -12,7 +14,7 @@ def main():
     console_scripts = ['protoc-gen-purerpc=purerpc.protoc_plugin.plugin:main']
     setup(
         name="purerpc",
-        version="0.1.6",
+        version=__version__,
         license="Apache License Version 2.0",
         description="Asynchronous pure python gRPC server and client implementation using curio "
                     "and hyper-h2.",
@@ -31,6 +33,7 @@ def main():
             "Operating System :: POSIX :: Linux",
             "Programming Language :: Python",
             "Programming Language :: Python :: 3 :: Only",
+            "Programming Language :: Python :: 3.5",
             "Programming Language :: Python :: 3.6",
             "Programming Language :: Python :: 3.7",
             "Programming Language :: Python :: Implementation :: CPython",
@@ -46,13 +49,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/_version.py

@@ -0,0 +1,3 @@
+# This file is imported from __init__.py and exec'd from setup.py
+
+__version__ = "0.2.0"

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/client.py

@@ -16,14 +16,12 @@ def __init__(self, host, port):
         self._host = host
         self._port = port
         self._grpc_socket = None
-        self._task_group = None
 
     async def __aenter__(self):
         await super().__aenter__()  # Does nothing
 
         background_task_group = await self.enter_async_context(anyio.create_task_group())
         self.push_async_callback(background_task_group.cancel_scope.cancel)
-        self._task_group = await self.enter_async_context(anyio.create_task_group())
 
         socket = await anyio.connect_tcp(self._host, self._port, autostart_tls=False, tls_standard_compatible=False)
         config = GRPCConfiguration(client_side=True)
@@ -57,10 +55,10 @@ def get_method_stub(self, method_name: str, signature: RPCSignature):
         stream_fn = functools.partial(self.rpc, method_name, signature.request_type,
                                       signature.response_type)
         if signature.cardinality == Cardinality.STREAM_STREAM:
-            return ClientStubStreamStream(stream_fn, self.channel._task_group)
+            return ClientStubStreamStream(stream_fn)
         elif signature.cardinality == Cardinality.UNARY_STREAM:
-            return ClientStubUnaryStream(stream_fn, self.channel._task_group)
+            return ClientStubUnaryStream(stream_fn)
         elif signature.cardinality == Cardinality.STREAM_UNARY:
-            return ClientStubStreamUnary(stream_fn, self.channel._task_group)
+            return ClientStubStreamUnary(stream_fn)
         else:
-            return ClientStubUnaryUnary(stream_fn, self.channel._task_group)
+            return ClientStubUnaryUnary(stream_fn)

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,
+        )