# Batching calls to server in python with asyncio

I've written this decorator to avoid too many requests to a backend. Did I get it right? Do I need some locks or can I assume some things are atomic / thread safe because this is asyncio?

def batch_calls(f):
"""Batches single call into one request.

Turns f, a function that gets a tuple of independent requests, into a function
that gets a single request.
"""
queue = {}

async def make_call():
if not queue:
return
await asyncio.sleep(0.1)
if not queue:
return
queue_copy = dict(queue)
queue.clear()
try:
for async_result, result in zip(
queue_copy.values(), await f(tuple(queue_copy))
):
async_result.set_result(result)
except Exception as exception:
for async_result in queue_copy.values():
async_result.set_exception(exception)

async def wrapped(hashable_input):
if hashable_input in queue:
return await queue[hashable_input]
async_result = asyncio.Future()
# Check again because of context switch due to the creation of asyncio.Future.
# TODO(uri): Make sure this is needed.
if hashable_input in queue:
return await queue[hashable_input]
queue[hashable_input] = async_result
if len(queue) == 1:
return await async_result

return wrapped


The user might use it like so:

@batch_calls
async def call_server(texts: Tuple[Text, ...]):
return tuple(
(
await httpx.post(
url=SERVICE_URL,
json={"texts": texts},
)
).json()
)


This code is causing a deadlock on some situations - not easily reproducible.

• Can you also post a more extended context: 1) a sample function f definition to which your decorator is applied; 2) sample hashable_input value ??? – RomanPerekhrest Dec 9 '19 at 5:57
• gave a run example. you can think of hashable_input as a string. ptal – Uri Dec 9 '19 at 11:03