aiobotocore-chrisglass 1.4.1.1

安装

$ pip install aiobotocore

基本示例

import asyncio
from aiobotocore.session import get_session

AWS_ACCESS_KEY_ID = "xxx"
AWS_SECRET_ACCESS_KEY = "xxx"


async def go():
    bucket = 'dataintake'
    filename = 'dummy.bin'
    folder = 'aiobotocore'
    key = '{}/{}'.format(folder, filename)

    session = get_session()
    async with session.create_client('s3', region_name='us-west-2',
                                   aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
                                   aws_access_key_id=AWS_ACCESS_KEY_ID) as client:
        # upload object to amazon s3
        data = b'\x01'*1024
        resp = await client.put_object(Bucket=bucket,
                                            Key=key,
                                            Body=data)
        print(resp)

        # getting s3 object properties of file we just uploaded
        resp = await client.get_object_acl(Bucket=bucket, Key=key)
        print(resp)

        # get object from s3
        response = await client.get_object(Bucket=bucket, Key=key)
        # this will ensure the connection is correctly re-used/closed
        async with response['Body'] as stream:
            assert await stream.read() == data

        # list s3 objects using paginator
        paginator = client.get_paginator('list_objects')
        async for result in paginator.paginate(Bucket=bucket, Prefix=folder):
            for c in result.get('Contents', []):
                print(c)

        # delete object from s3
        resp = await client.delete_object(Bucket=bucket, Key=key)
        print(resp)

loop = asyncio.get_event_loop()
loop.run_until_complete(go())

上下文管理器示例

from contextlib import AsyncExitStack

from aiobotocore.session import AioSession


# How to use in existing context manager
class Manager:
    def __init__(self):
        self._exit_stack = AsyncExitStack()
        self._s3_client = None

    async def __aenter__(self):
        session = AioSession()
        self._s3_client = await self._exit_stack.enter_async_context(session.create_client('s3'))

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

# How to use with an external exit_stack
async def create_s3_client(session: AioSession, exit_stack: AsyncExitStack):
    # Create client and add cleanup
    client = await exit_stack.enter_async_context(session.create_client('s3'))
    return client


async def non_manager_example():
    session = AioSession()

    async with AsyncExitStack() as exit_stack:
        s3_client = await create_s3_client(session, exit_stack)

        # do work with s3_client

支持的AWS服务

这是一个非详尽的列表，其中列出了aiobotocore针对AWS服务运行测试的内容。并非所有方法都经过测试，但我们旨在测试大多数常用方法。

由于boto3的实现方式，即使服务未在上文中列出，您也可以将任何放在await前面使其异步，例如，await client.list_named_queries()将异步列出所有命名的Athena查询。

如果某个服务未在此列出，并且您需要一些测试或示例，请随时提出问题。

运行测试

请确保您已安装开发要求，并且可以通过环境变量访问您的亚马逊密钥和密码。

$ cd aiobotocore
$ export AWS_ACCESS_KEY_ID=xxx
$ export AWS_SECRET_ACCESS_KEY=xxx
$ pipenv sync --dev

执行测试套件

$ py.test -v tests

邮件列表

https://groups.google.com/forum/#!forum/aio-libs

要求

• Python 3.6+

• aiohttp

• botocore

awscli

awscli依赖于botocore的单个版本，但是aiobotocore只支持特定范围的botocore版本。为确保您安装了最新版本的awscli，该版本与您特定的aiobotocore和botocore组合兼容，请使用

pip install -U aiobotocore[awscli]

更改