How to connect aioboto3 with yandex s3

Question

I have yandex s3 storage https://cloud.yandex.ru/ru/docs/storage/s3/?from=int-console-help-center-or-nav how to properly set up use of it with python aioboto3 lib

Answer 1

For me worked next code with use of aiobotocore

from aiobotocore.session import get_session
from io import BytesIO


AWS_ACCESS_KEY_ID = '<secret>'
AWS_SECRET_ACCESS_KEY = '<secret>'
AWS_S3_ENDPOINT_URL = 'https://storage.yandexcloud.net'
AWS_S3_REGION_NAME = 'storage'
BACKET_NAME = '<my_backet_name>'


READ_CHUNK_SIZE = 10 * 1024 * 1024  # 10 Mb


async def upload_file_to_s3(
    filename_key: str,
    file_stream: BytesIO,
):
    parts_info = []
    session = get_session()
    async with session.create_client(
        's3',
        region_name=AWS_S3_REGION_NAME,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
        aws_access_key_id=AWS_ACCESS_KEY_ID,
        endpoint_url=AWS_S3_ENDPOINT_URL,
    ) as client:
        create_multipart_upload_resp = await client.create_multipart_upload(
            Key=filename_key,
            Bucket=BACKET_NAME,
        )
        upload_id = create_multipart_upload_resp['UploadId']

        part_number = 0
        while True:
            contents = file_stream.read(READ_CHUNK_SIZE)
            if contents:
                part_number += 1
                part_upload_resp = await client.upload_part(
                    Body=contents,
                    UploadId=upload_id,
                    PartNumber=part_number,
                    Key=filename_key,
                    Bucket=BACKET_NAME,
                )
                parts_info.append(
                    {
                        'PartNumber': part_number,
                        'ETag': part_upload_resp['ETag']
                    }
                )
            else:
                break

        await client.complete_multipart_upload(
            UploadId=upload_id,
            Key=filename_key,
            Bucket=BACKET_NAME,
            MultipartUpload={"Parts": parts_info},
        )