Skip to content

Instantly share code, notes, and snippets.

@djosix
Last active August 13, 2019 18:17
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save djosix/58d260d3bcb5beddebe287aa964aef00 to your computer and use it in GitHub Desktop.
Save djosix/58d260d3bcb5beddebe287aa964aef00 to your computer and use it in GitHub Desktop.
import functools
import asyncio
import boto3
class AsyncBoto3Client:
def __init__(self, *args, **kwargs):
client = boto3.client(*args, **kwargs)
self.__client = client
for name in dir(client):
if name.startswith('_'):
continue
func = getattr(client, name)
if not callable(func):
continue
def wrapper(func, *args, **kwargs):
return asyncio.get_running_loop().run_in_executor(
None, lambda: func(*args, **kwargs))
setattr(self, name, functools.partial(wrapper, func))
async def main():
ecs = AsyncBoto3Client('ecs')
def wrap_task(task):
async def coro(task):
print(await task)
return asyncio.Task(coro(task))
future = asyncio.gather(
wrap_task(ecs.list_clusters()),
wrap_task(ecs.list_clusters()),
wrap_task(ecs.list_clusters()),
wrap_task(ecs.list_clusters()),
wrap_task(ecs.list_clusters()),
wrap_task(ecs.list_clusters()),
wrap_task(ecs.list_clusters()),
)
print('feels so good')
await future
asyncio.run(main())
@djosix
Copy link
Author

djosix commented Aug 13, 2019

very dirty solution

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment