storage_core

storage_core is an async-first, S3-compatible object storage abstraction library for Python. It provides a unified StorageClient interface, an S3StorageClient built on aiobotocore, and a fluent StorageBuilder API — making it the foundation for storing files, media, and binary data in the Dream ecosystem. It works with RustFS, MinIO, AWS S3, Ceph, and any other S3-compatible service.

Architecture

                        ┌─────────────────────────────────┐
                        │          storage_core           │
                        │   StorageBuilder (fluent API)   │
                        │      + MetadataSerializer       │
                        └───────────────┬─────────────────┘
                                        │ calls the client directly
                        ┌───────────────▼─────────────────┐
                        │      StorageClient (ABC)        │
                        └───────────────┬─────────────────┘
                                        │ implements
                        ┌───────────────▼─────────────────┐
                        │       S3StorageClient           │
                        │   (aiobotocore, lazy import)    │
                        └───────────────┬─────────────────┘
                                        │ S3 protocol
            ┌───────────────┬───────────┼───────────┬──────────────┐
            │               │           │           │              │
        RustFS           MinIO      AWS S3      Ceph        (any S3)

Class hierarchy

StorageClient (ABC)             ← 22 async methods: connection, buckets,
│                                  objects, presigned URLs, multipart
├── S3StorageClient             ← aiobotocore backend — default, works with any S3 service
└── (future backends)           ← implement the same ABC

StorageBuilder                  ← fluent entry point
├── BucketOperation             ← create, exists, delete, list_objects
├── ObjectOperation             ← put*, get, head, delete, copy_to, presigned_*
└── MultipartOperation          ← create, upload_part, complete, abort

MetadataSerializer (ABC)
├── JsonMetadataSerializer      ← default (non-strings → "json:"-prefixed JSON)
└── PassthroughMetadataSerializer

• Async-first: every I/O method is async def.
• Pluggable backends: one StorageClient ABC; unsupported methods raise NotImplementedError.
• Fluent builder: builder.object(bucket, key).put(data) — the builder calls the client directly, with no intermediate query object.
• Metadata-only serialization: object data is always raw bytes; only the metadata dict is serialized.

Install

As a standalone package

pip install .             # core — no mandatory dependencies
pip install ".[s3]"       # + aiobotocore for the S3 backend

Requires Python ≥ 3.11.

As a dependency from another service

storage_core is consumed via local path dependencies in the Dream workspace:

# requirements.txt — relative path to storage_core
../storage_core[s3]

# pyproject.toml
dependencies = [
    "storage_core[s3] @ file://../storage_core",
]

For local development and testing

pip install .
pip install -r requirements-dev.txt

Services consuming storage_core

All consumers install the package with the [s3] extra.

Importing

from storage_core import (
    StorageBuilder, StorageClient, S3StorageClient,
    create_s3_storage_client, get_s3_storage_client,
)

Available public exports

• StorageClient — abstract base class for all storage backends (22 async methods)
• S3StorageClient — aiobotocore-backed S3 client
• StorageBuilder — fluent entry point
• BucketOperation, ObjectOperation, MultipartOperation — builder operation classes
• StorageObject, ObjectListing — result dataclasses
• MetadataSerializer, JsonMetadataSerializer, PassthroughMetadataSerializer — metadata serializers
• create_s3_storage_client / get_s3_storage_client — factory functions (aliases of each other)

Quick start

Prerequisite: create_s3_storage_client() resolves credentials from env vars (RUSTFS_*, falling back to AWS_*). See Configuration.

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()
    await client.connect()

    async with StorageBuilder(client) as s:
        # ── Buckets ──────────────────────────────────────────────
        await s.bucket("dream-media").create()
        assert await s.bucket("dream-media").exists()

        # ── Put an object ────────────────────────────────────────
        obj = 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",
        )
        print(obj.key)        # "avatars/user-42.jpg"
        print(obj.metadata)   # {"user_id": "42", "source": "telegram"}

        # ── Get it back (response includes the size) ─────────────
        obj = await s.object("dream-media", "avatars/user-42.jpg").get()
        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 (lexicographic order) ───────
        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-2.jpg 1
        # avatars/user-42.jpg 22

        # ── 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.etag)  # e.g. '"2d9c1f4a8b6e..."'

    await client.disconnect()


asyncio.run(main())

The async with StorageBuilder(client) block only groups calls — it does not connect or disconnect the client. Manage the client lifecycle explicitly.

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 the extension (mimetypes)
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 upload — or abort it
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])
# await client.abort_multipart_upload("my-bucket", "big.bin", uid)

# Cleanup — S3 only deletes empty buckets
await client.delete_bucket("my-bucket")
await client.disconnect()

Configuration

Environment Variables (auto-resolved)

get_s3_storage_client is an alias for create_s3_storage_client — they behave identically.

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,   # optional — derived from the endpoint scheme when omitted
)

Metadata Serialization

Object data is always raw bytes. Only the metadata dict is serialized/deserialized via a MetadataSerializer. S3StorageClient uses JsonMetadataSerializer by default.

from storage_core import S3StorageClient, JsonMetadataSerializer

client = S3StorageClient(
    endpoint_url="http://localhost:9000",
    access_key="admin",
    secret_key="admin",
    metadata_serializer=JsonMetadataSerializer(),  # default — can be omitted
)

# 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)

StorageBuilder is also an async context manager, but __aexit__ does nothing — connect/disconnect the client explicitly.

BucketOperation

ObjectOperation

MultipartOperation

Testing

Run pytest from the package root — pytest.ini sets asyncio_mode = auto, so no @pytest.mark.asyncio decorators are needed. Unit tests run against FakeStorageClient from tests/conftest.py — no Docker or live S3 service required.

pip install -r requirements-dev.txt
pytest

Currently 31 tests across 3 files: test_builder.py (16), test_factory.py (7), test_serializer.py (8).

Project Layout

storage_core/
├── __init__.py          # Public API
├── storage_types.py     # StorageObject, ObjectListing
├── serializer.py        # MetadataSerializer (JSON, Passthrough)
├── builder.py           # StorageBuilder + Bucket/Object/Multipart operations
├── clients/
│   ├── base.py          # StorageClient ABC (22 methods)
│   ├── s3.py            # S3StorageClient (aiobotocore)
│   └── factory.py       # Env-var-based factories
├── tests/               # 31 tests across 3 files (FakeStorageClient fixture)
├── pyproject.toml       # Build config, optional deps: [s3]
└── requirements*.txt    # Core: none · Dev: pytest, pytest-asyncio, aiobotocore

Requirements

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

License

Proprietary — Dream ecosystem internal use.