void-ux / aiob2 Goto Github PK

import os
import asyncio

from aiob2 import Client

keys = [
    "some-big-file",
    "some-small-file"
]


async def download_file(client, bucket, key):
    print(f"About to download {bucket}/{key}")
    f = await client.download_file_by_name(
        file_name=key,
        bucket_name=bucket,
    )
    print(f"Downloaded {bucket}/{key}: {f.content_length} bytes")
    return f.content_length


async def main():
    print(f"Downloading objects {keys}")
    async with Client(os.environ["B2_APPLICATION_KEY_ID"], os.environ["B2_APPLICATION_KEY"]) as client:
        sizes = await asyncio.gather(*(download_file(client, "my-bucket", key) for key in keys))
    print(f"Object sizes: {sizes}")


if __name__ == '__main__':
    asyncio.run(main())

Downloading objects ['some-big-file', 'some-small-file']
About to download my-bucket/some-big-file
2023-05-04 13:17:15 INFO     aiob2.http Authenticating using static application key
About to download my-bucket/some-small-file
2023-05-04 13:17:15 INFO     aiob2.http Authenticating using static application key
Downloaded my-bucket/some-small-file: 6 bytes
Downloaded my-bucket/some-big-file: 15729152 bytes
Object sizes: ['15729152', '6']

        if (expires := headers.get('cache-control')) is not None:
            expires = datetime.datetime.strptime(expires, '%a, %d %b %Y %H:%M:%S %Z')
        else:
            expires = None
        self.expires: Optional[datetime.datetime] = expires

        self.cache_control: Optional[str] = headers.get('cache-control')
        if (expires := headers.get('expires')) is not None:
            expires = datetime.datetime.strptime(expires, '%a, %d %b %Y %H:%M:%S %Z')
        else:
            expires = None
        self.expires: Optional[datetime.datetime] = expires