Handtruck documentationΒΆ
An S3 client library built on HTTPX and AnyIO, compatible with any S3 provider.
from http import HTTPStatus
from httpx import AsyncClient
from handtruck import S3Client
client = S3Client(
url="https://s3.amazonaws.com",
client=AsyncClient(),
access_key_id="key-id",
secret_access_key="hackme",
region="us-east-1"
)
# Upload str object to bucket "bucket" and key "str"
resp = await client.put("bucket/str", "hello, world")
assert resp.status_code == HTTPStatus.OK
# Upload bytes object to bucket "bucket" and key "bytes"
resp = await client.put("bucket/bytes", b"hello, world")
assert resp.status_code == HTTPStatus.OK
# Upload AsyncIterable to bucket "bucket" and key "iterable"
async def gen():
yield b'some bytes'
resp = await client.put("bucket/file", gen())
assert resp.status_code == HTTPStatus.OK
# Upload file to bucket "bucket" and key "file"
resp = await client.put_file("bucket/file", "/path_to_file" )
assert resp.status_code == HTTPStatus.OK
# Check object exists using bucket+key
resp = await client.head("bucket/key")
assert resp.status_code == HTTPStatus.OK
# Get object by bucket+key
resp = await client.get("bucket/key")
data = resp.content
# Delete object using bucket+key
resp = await client.delete("bucket/key")
assert resp == HTTPStatus.NO_CONTENT
# List objects by prefix
async for result in client.list_objects_v2("bucket/", prefix="prefix"):
# Each result is a list of metadata objects representing an object
# stored in the bucket.
do_work(result)