Using aiobotocore with context managers

Question

I'm trying to use the aiobotocore library with context managers, but I'm having a hard time trying to configure my credentials. I need to create a class that configure my AWS client so I can use the put, read and delete functions in this library.

The following code is being used to this:

    from contextlib import AsyncExitStack
    from aiobotocore.session import AioSession
    from credentials import aws_access_key_id, aws_secret_access_key

    class AWSConnectionManager:

        def __init__(self, aws_acces_key_id, aws_secret_access_key):
            self.aws_acces_key_id = aws_acces_key_id
            self.aws_secret_access_key = aws_secret_access_key
            self._exit_stack = AsyncExitStack()
            self._client = None
            print('__init__')
             
        async def __aenter__(self):
            session = AioSession
            self._client = await self._exit_stack.enter_async_context(session.create_client('s3'))
            print('__aenter__')

        async def __aexit__(self, exc_type, exc_val, exc_tb):
            await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
            print('__aexit__')
            

    res = AWSConnectionManager(aws_acces_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key)

But, it doesn't pass through aenter and aexit method. With the code above I have the following output:


    __init__
    <__main__.AWSConnectionManager object at 0x7f03921ac640>

Does anyone know what can be wrong with my code?

user_6515755 · Answer

First: you need to fix 'session = AioSession' => 'session = AioSession()' + add return + pass credentials

async def __aenter__(self):
    session = AioSession()
    self._client = await self._exit_stack.enter_async_context(
        session.create_client(
            's3',
            aws_secret_access_key=self.aws_secret_access_key,
            aws_access_key_id=self.aws_access_key_id,
        )
    ) 
    return self

Second: you need to write/add proxy calls for put_object/get_object or make _client public by rename _client => client

 async def save_file(self, content, s3_filename: str):
     await self._client.put_object(Bucket=self.bucket, Body=content, Key=f'{s3_filename}')

 async def load_file(self, name):
     obj = await self._client.get_object(Bucket=self.bucket, Key=f'{name}')
     return obj['Body'].read()

now you can use like

async with SkyFileStorageProxy() as storage:
    await storage.load_file(name='test.txt')

Using aiobotocore with context managers

1 Answers