[ Python] aiobotocore를 사용한 AWS S3와 연동

Python으로 AWS와 연동할 때, boto3라이브러리를 사용하여 애플리케이션이나 스크립트를 Amazon S3, EC2, DynamoDB등의 서비스와 쉽게 연결하여 사용할 수 있습니다.

그 중 비동기적으로 파이썬 애플리케이션과 AWS 서비스를 연동하기 위해 asyncio 패키지와 boto3를 사용하여 패키징한 aiobotocore를 사용하여 비동기적으로 FastAPI와 AWS S3를 연동해보려 합니다. 이 때, aiobotocore를 사용할 수도 있지만, aioboto3를 사용할 수도 있습니다. 두 라이브러리의 차이점은 무엇일까요?

 

두 라이브러리의 공통점은 모두 Python에서 AWS 서비스에 접근할 수 있도록 해주는 비동기 라이브러리입니다.

하지만, aiobotocore는 비교적 저수준 라이브러리로 기본적인 API를 제공하며 더 많은 제어 기능을 갖고 있으며 이에 따라 더 높은 성능을 자랑합니다.

aioboto3는 aiobotocore API를 추상화하여 더 높은 수준의 API를 제공하게 되어 코드 작성이 더 간편하지만, 사용 가능한 기능이 제한될 수 있으며 이에 따라 성능 저하를 가져올 수 있습니다. 하지만 고수준 API를 제공하게 되면서 AWS 서비스별 고유 기능을 간편하게 지원할 수 있다는 장점이 있습니다.

 

즉, 성능이 중요하고 AWS 서비스에 대해 더 많은 제어가 필요한 경우에는 aiobotocore를 사용하고, 개발 속도가 중요하거나 AWS 서비스별 고유 기능을 사용해야 하며 더 많은 커뮤니티를 참고하고 싶은 경우 aioboto3를 사용하면 됩니다. 사용 목적에 따라 적합한 라이브러리를 선택하면 좋을 것 같습니다 ✔

 

references:
aiobotocore 문서: https://aiobotocore.readthedocs.io/en/latest/, https://github.com/aio-libs/aiobotocore
aioboto3 문서: https://aioboto3.readthedocs.io/en/latest/, https://github.com/terrycain/aioboto3

 

 

 

 

aiobotocore를 통해 AWS 서비스와 연동할 때, 자격 증명을 해야합니다.

해당 코드가 동작하는 애플리케이션이 AWS서비스에 접근할 수 있는 적절한 자격이 있는 지를 확인합니다. 이 때, 여러가지 방법으로 자격 증명을 할 수 있습니다. https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html#configuring-credentials

Credentials - Boto3 1.34.39 documentation

1. Passing credentials as parameters
2. Environment variables
3. Shared credentials file [ ~/.aws/credentials ]
4. AWS config file [ ~/.aws/config ]
5. Assume role provider  [ ~/.aws/config ]
6. Boto2 configuration file support  [ /etc/boto.cfg or ~/.boto ]
7. IAM roles

boto3 라이브러리는 위와 같은 순서로 AWS의 자격증명을 확인합니다. 순서대로 자격 증명을  확인하다가 찾는 죽시 확인 프로세스를 중지합니다. 다양한 방법이 있으니 개발 환경에 맞는 방법을 선택하면 좋을 것 같습니다.

저는 kubernetes상에 pod를 통해 python 애플리케이션을 배포할 예정으로, 해당 pod에 IRSA를 통해 AWS서비스에 접근하기 위한 권한을 부여하도록 kubernetes serviceaccount를 생성하고, 해당 serviceaccount에 IAM role을 attach하여 pod에서 해당 serviceaccount를 통해 자격 증명하도록 미리 사전에 구현하였습니다.

 

 

aiobotocore를 사용하여 AWS서비스에 연동하기 위한 방법도 여러가지가 있습니다.

client를 통해 API를 호출하게 되는데, 필요한 API호출 시 마다 aync with함수와 함께 aiobotocore session을 통해 client를 생성하여 사용하는 방법과 context manager형태로 client를 class로 생성하여 async with구문과 함께 사용되도록 하며,

with 구문에 들어갈 때와 나올 때의 동작을 __aenter_, __aexit__의 magic method로 정의하여 자원을 효율적으로 관리하게 할 수도 있습니다.

from contextlib import AsyncExitStack
from logging import getLogger

import aiobotocore.session
from fastapi import UploadFile
from types_aiobotocore_s3 import S3Client as Client
from types_aiobotocore_s3.literals import S3ServiceName
from types_aiobotocore_s3.type_defs import GetObjectOutputTypeDef
from typing_extensions import Self

from image_pool_manager.config import settings

SERVICE_NAME: S3ServiceName = "s3"
BUCKET_NAME: str = settings.s3_bucket_name
REGION_NAME: str = settings.region
CONTENT_TYPE: str = settings.content_type


class S3Client:
    def __init__(self) -> None:
        self.client: Client
        self.exit_stack = AsyncExitStack()

    async def __aenter__(self) -> Self:
        self.session = aiobotocore.session.get_session()
        self.client = await self.exit_stack.enter_async_context(
            self.session.create_client(
                service_name=SERVICE_NAME,
                region_name=REGION_NAME,
            )
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.exit_stack.__aexit__(exc_type, exc_val, exc_tb

context manager형태로 S3 client를 통해 AWS S3와 연동하도록 구현하였습니다.

aiobotocore의 session를 통해 AWS 서비스에 대한 연결 및 인증 정보를 관리하고, client를 생성하며 AWS의 서비스에 대한 요청을 처리하게 됩니다. 위와 같이 context manager를 통해 session과 client를 생성 및 사용이 끝나고 제거하도록 하여 자원을 효율적으로 관리할 수 있습니다.

client를 생성할 때, service name에는 s3나 dynamodb등이 사용될 수 있으며, aws credential key가 사용되지 않은 점은 위에서 언급한 IRSA를 통해 자격 증명이 이루어지기 때문에 따로 설정하지 않아도 됩니다.

 

async def upload_file(self, file: UploadFile, key: str) -> None:
    await self.client.upload_fileobj(
        Fileobj=file.file,
        Bucket=BUCKET_NAME,
        Key=key,
        ExtraArgs={"ContentType": CONTENT_TYPE},
    )

async def download_file(self, key: str) -> GetObjectOutputTypeDef:
    return await self.client.get_object(
        Bucket=BUCKET_NAME,
        Key=key,
    )

 

aiobotocore를 통해 client를 생성하고 나면, 위와 같이 s3에 file을 업로드 하거나, 다운로드 할 수 있습니다. 이외에도 dynamodb table을 생성하거나 해당 table의 CRUD를 수행할 수 있으며 AWS서비스에 대한 다양한 작업이 가능합니다.

위에서 작성한 upload_file, download_file을 통해 FastAPI 라우터에서 다음과 같이 API를 작성할 수 있습니다.

 

# upload API 
...

try:
    async with S3Client() as s3_client:
        await s3_client.upload_file(file=file, key=str(file.filename))
except botocore.exceptions.ClientError as err:
    logger.error(
        f"An error occurred while uploading the \
            file[{file.filename}] on the S3."
    )
    raise HTTPException(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        detail=err.response["Error"],
    )

try:
    async with DBClient() as db_client:
        await db_client.put_item(metadata=metadata)
except botocore.exceptions.ClientError as err:
    logger.error(
        f"An error occurred while uploading the metadata[{metadata}] on the dynamodb."
    )
    raise HTTPException(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        detail=err.response["Error"],
    )

return {"file_size": file.size}

# donwload API
...

try:
    async with S3Client() as s3_client:
        result = await s3_client.download_file(key=key)
except botocore.exceptions.ClientError as err:
    raise HTTPException(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        detail=err.response["Error"],
    )
return StreamingResponse(
    result["Body"], media_type=settings.media_type
)

 

이렇게 aiobotocore를 통해 AWS 서비스와 연동하는 법에 대해 공유해봤습니다.
해당 내용에 대해 조언해주시거나 다양한 피드백은 언제나 환영합니다. 감사합니다 🙌