python-betterproto
python-betterproto copied to clipboard
danielgtaylor
Option for generating code without async?
I have some infrastructure developed around the previously gen'd grpc code, however I would really like to adopt betterproto for all its extra pythonic features.
The only hurdle i'm facing is that the async implementation is turned on by default. Is there a way to turn this off and not have the async gen code?
Just to clarify beyond all doubt here -- for the moment the answer here is no, correct? We can't turn off async/await in the generated stub?
This library does not work for Cloud Run due to the lack of this functionality. If it worked for sure betterproto would be the best option.
What about using native Python's async-to-sync functionality? That is, given an async RPC method called via result = await ServiceStub.some_method(), you could do either:
result = asyncio.run(ServiceStub.some_method())
...if you wanted to start and stop the event loop just for that method. If that incurs too much overhead, you could call into an existing event loop via something like:
result = asyncio.run_coroutine_threadsafe(ServiceStub.some_method(), loop=asyncio.get_event_loop()).result()
@nat-n if that satisfies user requirements, would you accept a PR documenting that technique?
I tried that as a workaround. The problem was that
- grpclib channel constructor requires a running working loop. Constructor can't be marked as async so using those run in async functions wouldn't work. So you'd have to set a main loop before calling that. I'm not sure about the impact of doing so in asyncio-based web servers like gunicorn w/aiohttp or uvicorn. From running in production for the last quarter I never had any issue.
- That loop is used to maintain connection keepalive. Running RPC and the event loop to RPC completion works until the remote end timeout the connection (because no loop is running to reply to pings). I believe normal gRPC library has automatic retry in this case but grpclib would just stop working entirely.
- One workaround I use is to call
channel.close()after every RPC to always get a fresh connection, but that's not efficient at all.
If I may suggest, protobuf-ts has a swappable transport implementation for RPC where you can swap between gRPC, twirp and grpcweb.
I have a similar requirement for a client; and I thought I'd share my current workaround. I didn't want to edit the generated client stubs because I may need to re-generate them frequently. I also wanted to hide the async stuff from the user, and let them call the service methods as if they're synchronous.
To achieve this, I created two decorators.
import asyncio
from functools import wraps
def synced(async_func):
"""Wrap the asynchronous function so that it becomes synchronous"""
@wraps(async_func)
def wrapped(*args, **kwargs):
return asyncio.get_event_loop().run_until_complete(async_func(*args, **kwargs))
return wrapped
def sync(cls):
"""Decorator for the service stubs to make them synchronized"""
# Find all service methods (endpoints) by inspecting the passed class
endpoints = tuple(
getattr(cls, endpoint)
for endpoint in dir(cls)
if callable(getattr(cls, endpoint)) and not endpoint.startswith("_")
)
# Replace the methods by their `synced` variants
for endpoint in endpoints:
setattr(cls, endpoint.__name__, synced(endpoint))
return cls
The decorator synced creates a single async event loop each time the method is called and runs the loop until the method is complete. The decorator sync is then used to apply the synced decorator to each service method residing in the service stub class. it assumes that the service methods are callables that don't start with an underscore.
To get a synchronized service from the generated asynchronous ServiceStub, you initiate it as follows:
synchronized_service = sync(ServiceStub(channel=...)) # enter a channel as you normally would
The upside of this method is that your IDE will pick up the service methods. The downside of this method is that tools like mypy get confused because by applying these decorators, we are (dynamically) changing the return types.
@wpdonders that solution works, but may break in surprising ways in the presence of multiprocessing or other threads. I suggest using the asgiref.sync package (specifically async_to_sync), which provides the same functionality without those drawbacks.