asyncio boto3 wrapper to run boto3 client methods in a thread pool executor as an alternative to aiobotocore

boto3async.py

# =============================================================================
# Copyright 2022 by Rafid Al-Humaimidi. All rights reserved.
# Licensed via Apache 2.0 (https://www.apache.org/licenses/LICENSE-2.0)
#
# Forked from: https://github.com/rafidka/boto3async
# =============================================================================

"""Adds simple async wrappers around boto3 client methods.

This module adds async methods to the stock boto3 clients. The async versions of
each method are suffixed with "_async" rather than overriding the regular
methods in-place.

The async methods simply wrap the blocking methods and run them in the default
thread pool executor via asyncio.to_thread().

This strategy ensures feature parity with boto3 and good-enough-performance
for most use cases where the the additional complexity of aiobotocore is
not worth the tradeoff.

Example::

    import asyncio

    import boto3async

    async def main():
        s3 = boto3async.client('s3')
        response = await s3.list_buckets_async()
        print(response)


    asyncio.run(main())

"""

import asyncio
import boto3
import re
import functools
import contextvars


# From https://stackoverflow.com/a/1176023/196697
def _camel_to_snake(name):
    """
    Convert a name in camel case to snake case.

    Arguments:
    name -- The name to convert.

    Returns:
    The name in snake case.
    """
    name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
    return re.sub('([a-z0-9])([A-Z])', r'\1_\2', name).lower()


def asyncify_client(client):
    """
    Adds async methods to each of the sync methods of a boto3 client.

    Keyword arguments
    client -- The client to add sync methods to. Notice that the client
        will be updated in place, and will also be returned as a return
        value.

    Returns:
    The same client.
    """

    def create_async_func(sync_func):
        async def async_func(*args, **kwargs):
            return await asyncio.to_thread(sync_func, *args, **kwargs)
        return async_func

    for operation in client._service_model.operation_names:
        operation_camelcase = _camel_to_snake(operation)
        sync_func = getattr(client, operation_camelcase)
        async_func = create_async_func(sync_func)
        setattr(client, f'{operation_camelcase}_async', async_func)

    return client


def client(*args, **kwargs):
    """
    Create a normal boto3 client and add async methods to it.

    Arguments:
    See boto3 documentation for what arguments you need to pass to create the
    different AWS client.

    Returns:
    A boto3 client with async versions of each method. Each async method has
    the same name as the sync method, but with "_async" postfix. For example,
    the async version of list_buckets is named list_buckets_async.
    """
    client = boto3.client(*args, **kwargs)

    return asyncify_client(client)


def resource(*args, **kwargs):
    return boto3.resource(*args, **kwargs)

Note: This could be modified to use a dedicated executor if desired.