Как оптимизировать асинхронное загрузку видео на S3 Python? ⇐ Python

[Anonymous]

Я делаю запросы на различные конечные точки API, которые содержат большие файлы MP4 (> 100 МБ) и хочу заполнить их до ведра S3. Я заключил это как задачу ввода/вывода из -за обработки запросов API + потоковую передачу видеоконтента для загрузки, поэтому я решил использовать асинхронное программирование. Моя текущая реализация кода не достаточно эффективна, когда дело доходит до загрузки на S3. Как я приблизился к тому, что это было сначала итерация над каждым куском видео, а затем загрузила все это, но понял, что это заняло несколько минут на видео. Затем я пробовал загрузить каждый итерационный кусок, похожий на эту реализацию, но не видел никаких улучшений. Я почти уверен, что это проблема с тем, как я обращаюсь с каждой частью видеопотока, но не нашел много решений, чтобы помочь. Я могу загрузить видео в свой локальный каталог с каждым видео, показывающим и обработка по задачам, но я заметил на S3, похоже, что видео обрабатываются по одному. Я хотел проверить, является ли этот код асинхронным? Похоже, это не для меня из -за того, сколько времени требуется для загрузки на S3. Если это не так, как я могу сделать это асинхронным?import asyncio
import httpx
from botocore.exceptions import ClientError
import aioboto3
from typing import AsyncIterator

async def async_upload(stream: AsyncIterator[bytes],
destination_key: str,
) -> None:
"""
Asynchronously uploads video files via `aioboto3`

Args:
stream (AsyncIterator[bytes]): The video stream broken up into iterable chunks
destination_key (str): The desired destination for the video file
"""

session = aioboto3.Session()

async with session.client('s3') as s3:
resp = await s3.create_multipart_upload(Bucket=bucket_name, Key=destination_key)

upload_id = resp['UploadId']

parts = []
p_number = 1

try:
async for chunk in stream:
part_resp = await s3.upload_part(
Bucket=bucket_name,
Key=destination_key,
PartNumber=p_number,
UploadId=upload_id,
Body=chunk
)
parts.append({
'ETag': part_resp['ETag'],
'PartNumber': p_number
})

p_number += 1
print('Chunk done')

await s3.complete_multipart_upload(
Bucket=bucket_name,
Key=destination_key,
UploadId=upload_id,
MultipartUpload={'Parts': parts}
)
print('Upload Done')
except Exception as e:
await s3.abort_multipart_upload(
Bucket=bucket_name,
Key=destination_key,
UploadId=upload_id
)
print('An error occured', e)

async def upload_video(url):
# Example url
async with httpx.AsyncClient(timeout=httpx.Timeout(20, read=10)) as request:
# Stream to lazily load to memory for faster times
async with request.stream("GET", url) as response:
if response.status_code == 200:
try:
s3_key = f'{folder_name}/{file_name}'
await async_upload(response.aiter_bytes(chunk_size=100*1024*1024), s3_key) # 100MB chunks

except ClientError as e:
print('Upload Failed', e)
else:
print('Bad Response', response.status_code)

async def main():
urls = ['https://api.example.com/somevideo.mp4', 'https://api.example.com/somevideo.mp4',
'https://api.example.com/somevideo.mp4']

tasks = [
upload_video(url)
for url in urls
]

await asyncio.gather(*tasks)

if __name__ == '__main__':

bucket_name = 'Test-Bucket'
folder_name = 'Test'
file_name = 'Video.MP4'

asyncio.run(main())


Подробнее здесь: https://stackoverflow.com/questions/796 ... -s3-python