storage_core

Async-first S3-compatible object storage abstraction for the Dream ecosystem.
Uses aiobotocore under the hood — works with RustFS, MinIO, AWS S3, Ceph, and any S3-compatible service.

Quickstart

import asyncio
from storage_core import StorageBuilder, create_s3_storage_client


async def main():
    # Create a client — auto-resolves RUSTFS_* env vars, falls back to AWS_*
    client = create_s3_storage_client()

    async with StorageBuilder(client) as s:
        await s.connect()

        # ── Buckets ──────────────────────────────────────────────
        await s.bucket("dream-media").create()
        assert await s.bucket("dream-media").exists()

        # ── Put an object ────────────────────────────────────────
        await s.object("dream-media", "avatars/user-42.jpg").put(
            b"binary-image-data-here",
            metadata={"user_id": "42", "source": "telegram"},
            content_type="image/jpeg",
        )

        # ── Get it back ──────────────────────────────────────────
        obj = await s.object("dream-media", "avatars/user-42.jpg").get()
        print(obj.key)        # "avatars/user-42.jpg"
        print(obj.size)       # 22
        print(obj.metadata)   # {"user_id": "42", "source": "telegram"}
        print(obj.data[:10])  # b"binary-ima"

        # ── Head (metadata only, no data) ────────────────────────
        meta = await s.object("dream-media", "avatars/user-42.jpg").head()
        print(meta.size)      # 22
        print(meta.data)      # None

        # ── List objects with prefix ─────────────────────────────
        await s.object("dream-media", "avatars/user-1.jpg").put(b"a")
        await s.object("dream-media", "avatars/user-2.jpg").put(b"b")
        await s.object("dream-media", "docs/readme.md").put(b"# doc")

        listing = await s.bucket("dream-media").list_objects(prefix="avatars/")
        for o in listing.objects:
            print(o.key, o.size)
        # avatars/user-1.jpg 1
        # avatars/user-42.jpg 22
        # avatars/user-2.jpg 1

        # ── Copy ─────────────────────────────────────────────────
        await s.object("dream-media", "avatars/user-42.jpg").copy_to(
            "dream-media", "backup/user-42.jpg"
        )

        # ── Delete ───────────────────────────────────────────────
        await s.object("dream-media", "avatars/user-1.jpg").delete()

        # ── Presigned URLs ───────────────────────────────────────
        download_url = await s.object(
            "dream-media", "avatars/user-42.jpg"
        ).presigned_get(expires_in=3600)
        print(download_url)
        # http://localhost:9000/dream-media/avatars/user-42.jpg?X-Amz-...

        upload_url = await s.object(
            "dream-media", "uploads/new-file.bin"
        ).presigned_put(expires_in=600)
        # Client can PUT to this URL directly

        # ── Multipart upload (large files) ───────────────────────
        mp = s.multipart("dream-media", "large-archive.tar")
        upload_id = await mp.create(
            metadata={"compressed": "true"},
            content_type="application/x-tar",
        )
        part1 = await mp.upload_part(upload_id, 1, b"chunk-A" * 1000)
        part2 = await mp.upload_part(upload_id, 2, b"chunk-B" * 1000)
        result = await mp.complete(upload_id, [part1, part2])
        print(result.size)  # 14000

        await s.disconnect()


asyncio.run(main())

Put Sources

put() is shorthand for put_from_data(bytes). Three dedicated methods let you upload from different sources:

# Raw bytes
await client.put_from_data("bucket", "key", b"hello", metadata={"kind": "text"})

# Local file — content_type auto-detected from extension
await client.put_from_path("bucket", "photos/avatar.png", "/tmp/avatar.png")

# Async stream / async iterable
import aiofiles
async with aiofiles.open("/tmp/video.mp4", "rb") as f:
    await client.put_from_source("bucket", "videos/video.mp4", f,
                                 content_type="video/mp4")

Builder equivalents:

await s.object("b", "k").put(b"hello")                        # bytes
await s.object("b", "k").put_from_data(b"hello")              # identical
await s.object("b", "k").put_from_path("/tmp/avatar.png")     # file
await s.object("b", "k").put_from_source(async_stream)        # stream

Direct Client Usage (without Builder)

from storage_core import S3StorageClient

client = S3StorageClient(
    endpoint_url="http://localhost:9000",
    access_key="rustfsadmin",
    secret_key="rustfsadmin",
)
await client.connect()

# Bucket management
await client.create_bucket("my-bucket")
buckets = await client.list_buckets()        # ["my-bucket"]
exists = await client.bucket_exists("my-bucket")  # True

# Object CRUD — put (shorthand for put_from_data)
obj = await client.put("my-bucket", "data.json", b'{"x":1}',
                       metadata={"version": "1.0"})

# Or use the dedicated methods
obj = await client.put_from_data("my-bucket", "data.json", b'{"x":1}')
obj = await client.put_from_path("my-bucket", "data.json", "/tmp/data.json")
obj = await client.put_from_source("my-bucket", "data.json", async_stream)

obj = await client.get("my-bucket", "data.json")
meta = await client.head("my-bucket", "data.json")
await client.delete("my-bucket", "data.json")

# List with pagination
page1 = await client.list_objects("my-bucket", prefix="logs/", max_keys=100)
if page1.is_truncated:
    page2 = await client.list_objects("my-bucket", prefix="logs/",
                                       max_keys=100, token=page1.next_token)

# Copy
await client.copy("my-bucket", "src.txt", "my-bucket", "dst.txt")

# Presigned URLs
url = await client.presign_get("my-bucket", "file.pdf", expires_in=3600)

# Multipart
uid = await client.create_multipart_upload("my-bucket", "big.bin")
p1 = await client.upload_part("my-bucket", "big.bin", uid, 1, b"part1")
p2 = await client.upload_part("my-bucket", "big.bin", uid, 2, b"part2")
result = await client.complete_multipart_upload("my-bucket", "big.bin", uid, [p1, p2])

# Cleanup
await client.delete_bucket("my-bucket")
await client.disconnect()

Configuration

Environment Variables (auto-resolved)

Development (docker-compose)

# infrastructure/development/.env
RUSTFS_ENDPOINT_URL=http://localhost:9000
RUSTFS_ACCESS_KEY=rustfsadmin
RUSTFS_SECRET_KEY=rustfsadmin

Then:

from storage_core import create_s3_storage_client

client = create_s3_storage_client()  # reads env vars automatically
await client.connect()
assert await client.ping()  # True

Explicit Configuration

from storage_core import S3StorageClient

client = S3StorageClient(
    endpoint_url="https://s3.amazonaws.com",
    access_key="AKIAIOSFODNN7EXAMPLE",
    secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    region="us-west-2",
    secure=True,
)

Metadata Serialization

Object data is always raw bytes. The optional MetadataSerializer handles the metadata dict only.

from storage_core import JsonMetadataSerializer

client = S3StorageClient(
    endpoint_url="http://localhost:9000",
    access_key="admin",
    secret_key="admin",
    metadata_serializer=JsonMetadataSerializer(),
)

# Non-string values are JSON-encoded with a "json:" prefix
await client.put("bucket", "obj", b"data", metadata={
    "author": "dream",      # string → stored as-is
    "count": 42,            # int → stored as "json:42"
    "tags": ["a", "b"],     # list → stored as 'json:["a","b"]'
})

obj = await client.get("bucket", "obj")
print(obj.metadata)  # {"author": "dream", "count": 42, "tags": ["a", "b"]}

Available serializers:

Builder API Reference

StorageBuilder(client)

BucketOperation

ObjectOperation

MultipartOperation

Architecture

storage_core/
├── __init__.py          # Public API
├── storage_types.py     # StorageObject, ObjectListing
├── serializer.py        # MetadataSerializer (JSON, Passthrough)
├── builder.py           # StorageBuilder + operation sub-builders
└── clients/
    ├── base.py          # StorageClient ABC (21 methods)
    ├── s3.py            # S3StorageClient (aiobotocore)
    └── factory.py       # Env-var-based factories

Requirements

• Python ≥ 3.11
• aiobotocore (installed automatically with pip install storage_core[s3])

License

Proprietary — Dream ecosystem internal use.