aioaws

为某些AWS服务提供的Asyncio SDK。

这个库不依赖于boto, boto3或其他任何臃肿、不透明且难以理解的AWS SDK。相反，它从头开始编写，以提供对我想使用的AWS服务的干净、安全且易于调试的访问。

库使用black格式化，并包含完整的类型提示（mypy在严格模式下通过）。

它目前支持

• S3 - 列表、删除、递归删除、生成签名上传URL、生成签名下载URL
• SES - 发送电子邮件，包括带有附件的多部分邮件
• SNS - 足以从SES接收邮件投递通知
• AWS Signature Version 4 身份验证适用于任何 AWS 服务（这是我所知的 Python 中 AWS4 的唯一干净且现代实现，请参阅 core.py）

aioaws 的唯一依赖项是

• aiofiles - 用于异步读取文件
• cryptography - 用于验证 SNS 签名
• httpx - 用于 HTTP 请求
• pydantic - 用于验证响应

安装

pip install aioaws

S3 使用

import asyncio
from aioaws.s3 import S3Client, S3Config
from httpx import AsyncClient

# requires `pip install devtools`
from devtools import debug

async def s3_demo(client: AsyncClient):
    s3 = S3Client(client, S3Config('<access key>', '<secret key>', '<region>', 'my_bucket_name.com'))

    # upload a file:
    await s3.upload('path/to/upload-to.txt', b'this the content')

    # list all files in a bucket
    files = [f async for f in s3.list()]
    debug(files)
    """
    [
        S3File(
            key='path/to/upload-to.txt',
            last_modified=datetime.datetime(...),
            size=16,
            e_tag='...',
            storage_class='STANDARD',
        ),
    ]
    """
    # list all files with a given prefix in a bucket
    files = [f async for f in s3.list('path/to/')]
    debug(files)

    # # delete a file
    # await s3.delete('path/to/file.txt')
    # # delete two files
    # await s3.delete('path/to/file1.txt', 'path/to/file2.txt')
    # delete recursively based on a prefix
    await s3.delete_recursive('path/to/')

    # generate an upload link suitable for sending to a browser to enabled
    # secure direct file upload (see below)
    upload_data = s3.signed_upload_url(
        path='path/to/',
        filename='demo.png',
        content_type='image/png',
        size=123,
    )
    debug(upload_data)
    """
    {
        'url': 'https://my_bucket_name.com/',
        'fields': {
            'Key': 'path/to/demo.png',
            'Content-Type': 'image/png',
            'AWSAccessKeyId': '<access key>',
            'Content-Disposition': 'attachment; filename="demo.png"',
            'Policy': '...',
            'Signature': '...',
        },
    }
    """

    # generate a temporary link to allow yourself or a client to download a file
    download_url = s3.signed_download_url('path/to/demo.png', max_age=60)
    print(download_url)
    #> https://my_bucket_name.com/path/to/demo.png?....

async def main():
    async with AsyncClient(timeout=30) as client:
        await s3_demo(client)

asyncio.run(main())

上例中显示的 upload_data 可以在 JS 中使用类似以下的方式

const formData = new FormData()
for (let [name, value] of Object.entries(upload_data.fields)) {
  formData.append(name, value)
}
const fileField = document.querySelector('input[type="file"]')
formData.append('file', fileField.files[0])

const response = await fetch(upload_data.url, {method: 'POST', body: formData})

(在获取 upload_data 的请求中，您需要提供文件大小和内容类型，以便使此处显示的上传成功)

SES

使用 SES 发送电子邮件

import asyncio
from pathlib import Path
from httpx import AsyncClient
from aioaws.ses import SesConfig, SesClient, SesRecipient, SesAttachment

async def ses_demo(client: AsyncClient):
    ses_client = SesClient(client, SesConfig('<access key>', '<secret key>', '<region>'))

    message_id = await ses_client.send_email(
        SesRecipient('sende@example.com', 'Sender', 'Name'),
        'This is the subject',
        [SesRecipient('recipient@eample.com', 'John', 'Doe')],
        'this is the plain text body',
        html_body='<b>This is the HTML body.<b>',
        bcc=[SesRecipient(...)],
        attachments=[
            SesAttachment(b'this is content', 'attachment-name.txt', 'text/plain'),
            SesAttachment(Path('foobar.png')),
        ],
        unsubscribe_link='https:://example.com/unsubscribe',
        configuration_set='SES configuration set',
        message_tags={'ses': 'tags', 'go': 'here'},
    )
    print('SES message ID:', message_id)

async def main():
    async with AsyncClient() as client:
        await ses_demo(client)

asyncio.run(main())

SNS

从 SNS 接收有关 SES Webhooks 的数据（假设您正在使用 FastAPI）

from aioaws.ses import SesWebhookInfo
from aioaws.sns import SnsWebhookError
from fastapi import Request
from httpx import AsyncClient

async_client = AsyncClient...

@app.post('/ses-webhook/', include_in_schema=False)
async def ses_webhook(request: Request):
    request_body = await request.body()
    try:
        webhook_info = await SesWebhookInfo.build(request_body, async_client)
    except SnsWebhookError as e:
        debug(message=e.message, details=e.details, headers=e.headers)
        raise ...

    debug(webhook_info)
    ...

有关 SesWebhookInfo 中提供的更多信息，请参阅 此处。