Use aiobotocore 2.2.0 to support assume role credentials

setup.py

@@ -58,15 +58,17 @@ def readme():
     install_requires=[
         'python-dateutil>=2.8',
         'thumbor>=7.0.0a2,<8',
-        'aiobotocore==0.12.0',
-        'boto3>=1.9,<1.13',
+        'aiobotocore==2.2.0',
+        'botocore>=1.24.21,<1.24.22',
+        'boto3>=1.21.21,<1.21.22'
     ],
     extras_require={
         'tests': [
             'coverage>=6.5',
             'moto[server]>=4.0',
             'mock>=4.0',
             'pytest>=7.2',
+            'pytest-asyncio>=0.20.2'
         ],
     },
 )

tc_aws/aws/bucket.py

@@ -3,8 +3,9 @@
 # Copyright (c) 2015, thumbor-community
 # Use of this source code is governed by the MIT license that can be
 # found in the LICENSE file.
+import io
 
-import aiobotocore
+import aiobotocore.session
 from botocore.client import Config
 from thumbor.utils import logger
 from thumbor.engines import BaseEngine
@@ -26,6 +27,7 @@ def __new__(cls, bucket, region, endpoint, *args, **kwargs):
     """
     This handles all communication with AWS API
     """
+
     def __init__(self, bucket, region, endpoint, max_retry=None):
         """
         Constructor
@@ -44,24 +46,22 @@ def __init__(self, bucket, region, endpoint, max_retry=None):
                 )
             )
 
-        if self._client is None:
-            self._client = aiobotocore.get_session().create_client(
-                's3',
-                region_name=region,
-                endpoint_url=endpoint,
-                config=config
-            )
+        self.region_name = region
+        self.endpoint_url = endpoint
+        self.session = aiobotocore.session.get_session()
 
     async def exists(self, path):
         """
         Checks if an object exists at a given path
         :param string path: Path or 'key' to retrieve AWS object
         """
         try:
-            await self._client.head_object(
-                Bucket=self._bucket,
-                Key=self._clean_key(path),
-            )
+            async with self.session.create_client('s3', region_name=self.region_name,
+                                                  endpoint_url=self.endpoint_url) as s3_client:
+                await s3_client.head_object(
+                    Bucket=self._bucket,
+                    Key=self._clean_key(path),
+                )
         except Exception:
             return False
         return True
@@ -71,11 +71,16 @@ async def get(self, path):
         Returns object at given path
         :param string path: Path or 'key' to retrieve AWS object
         """
+        async with self.session.create_client('s3', region_name=self.region_name,
+                                              endpoint_url=self.endpoint_url) as s3_client:
+            response = await s3_client.get_object(Bucket=self._bucket, Key=self._clean_key(path))
+            # TODO: Verify if it is possible to restore the original behavior were response['Body'] was a coroutine.
+            # Thumbor was getting stuck when response['Body'].read() was being called by s3_loader.load.
+            # To Avoid this, we read the Body content and expose it as a BytesIO to maintain the interface.
+            content = await response['Body'].read()
+            response['Body'] = io.BytesIO(content)
 
-        return await self._client.get_object(
-            Bucket=self._bucket,
-            Key=self._clean_key(path),
-        )
+        return response
 
     async def get_url(self, path, method='GET', expiry=3600):
         """
@@ -84,18 +89,19 @@ async def get_url(self, path, method='GET', expiry=3600):
         :param string method: Method for requested URL
         :param int expiry: URL validity time
         """
+        async with self.session.create_client('s3', region_name=self.region_name,
+                                              endpoint_url=self.endpoint_url) as s3_client:
+            url = await s3_client.generate_presigned_url(
+                ClientMethod='get_object',
+                Params={
+                    'Bucket': self._bucket,
+                    'Key': self._clean_key(path),
+                },
+                ExpiresIn=expiry,
+                HttpMethod=method,
+            )
 
-        url = await self._client.generate_presigned_url(
-            ClientMethod='get_object',
-            Params={
-                'Bucket': self._bucket,
-                'Key': self._clean_key(path),
-            },
-            ExpiresIn=expiry,
-            HttpMethod=method,
-        )
-
-        return url
+            return url
 
     async def put(self, path, data, metadata=None, reduced_redundancy=False, encrypt_key=False):
         """
@@ -123,17 +129,21 @@ async def put(self, path, data, metadata=None, reduced_redundancy=False, encrypt
         if metadata is not None:
             args['Metadata'] = metadata
 
-        return await self._client.put_object(**args)
+        async with self.session.create_client('s3', region_name=self.region_name,
+                                              endpoint_url=self.endpoint_url) as s3_client:
+            return await s3_client.put_object(**args)
 
     async def delete(self, path):
         """
         Deletes key at given path
         :param string path: Path or 'key' to delete
         """
-        return await self._client.delete_object(
-            Bucket=self._bucket,
-            Key=self._clean_key(path),
-        )
+        async with self.session.create_client('s3', region_name=self.region_name,
+                                              endpoint_url=self.endpoint_url) as s3_client:
+            return await s3_client.delete_object(
+                Bucket=self._bucket,
+                Key=self._clean_key(path),
+            )
 
     def _clean_key(self, path):
         logger.debug('Cleaning key: {path!r}'.format(path=path))

tc_aws/loaders/s3_loader.py

@@ -63,8 +63,9 @@ async def load(context, url):
         return result
 
     result.successful = True
-    async with file_key['Body'] as stream:
-        result.buffer = await stream.read()
+    with file_key['Body'] as stream:
+        result.buffer = stream.read()
+        del stream
 
     result.metadata.update(
         size=file_key['ContentLength'],

tc_aws/result_storages/s3_storage.py

@@ -60,8 +60,10 @@ async def get(self, path = None):
             return None
 
         result = ResultStorageResult()
-        async with key['Body'] as stream:
-            result.buffer = await stream.read()
+        with key['Body'] as stream:
+            result.buffer = stream.read()
+            del stream
+
         result.successful = True
 
         result.metadata = {

tc_aws/storages/s3_storage.py

@@ -109,8 +109,9 @@ async def get_crypto(self, path):
             logger.warn("[STORAGE] s3 key not found at %s" % crypto_path)
             return None
 
-        async with file_key['Body'] as stream:
-            file_key = await stream.read()
+        with file_key['Body'] as stream:
+            file_key = stream.read()
+            del stream
 
         return file_key.decode('utf-8')
 
@@ -130,8 +131,8 @@ async def get_detector_data(self, path):
         if not file_key or self.is_expired(file_key) or 'Body' not in file_key:
             return None
 
-        async with file_key['Body'] as stream:
-            return loads(await stream.read())
+        with file_key['Body'] as stream:
+            return loads(stream.read())
 
     async def get(self, path):
         """
@@ -144,8 +145,8 @@ async def get(self, path):
         except BotoCoreError:
             return None
 
-        async with file['Body'] as stream:
-            return await stream.read()
+        with file['Body'] as stream:
+            return stream.read()
 
     async def exists(self, path):
         """