Metadata-Version: 2.4
Name: storage_core
Version: 0.1.1
Summary: Async-first S3-compatible object storage abstraction layer with pluggable backends.
Project-URL: homepage, https://example.com/storage_core
Project-URL: repository, https://example.com/storage_core.git
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Provides-Extra: s3
Requires-Dist: aiobotocore>=2.0; extra == "s3"

# 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

```bash
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:

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

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

### For local development and testing

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

## Services consuming storage_core

| Service                               | How it integrates                                                                |
| ------------------------------------- | -------------------------------------------------------------------------------- |
| **web-downloader** (`crawlers/web/`)  | Uploads downloaded media to RustFS via `RustFSUploader` (uses `StorageBuilder`). |
| **tele-media** (`crawlers/telegram/`) | Uploads Telegram media files through its `MediaUploader` wrapper.                |
| **codal-crawler** (`crawlers/web/`)   | Stores attachment files (PDF, Excel, images) in RustFS.                          |
| **codal-parser** (`crawlers/web/`)    | Uses `create_s3_storage_client()` for attachment/file management.                |

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

## Importing

```python
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](#configuration).

```python
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:

```python
# 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:

```python
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)

```python
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)

| Variable              | Fallback                           | Default      |
| --------------------- | ---------------------------------- | ------------ |
| `RUSTFS_ENDPOINT_URL` | `S3_ENDPOINT_URL`                  | _(required)_ |
| `RUSTFS_ACCESS_KEY`   | `AWS_ACCESS_KEY_ID`                | _(required)_ |
| `RUSTFS_SECRET_KEY`   | `AWS_SECRET_ACCESS_KEY`            | _(required)_ |
| `RUSTFS_REGION`       | `S3_REGION` → `AWS_DEFAULT_REGION` | `us-east-1`  |

`get_s3_storage_client` is an alias for `create_s3_storage_client` — they behave identically.

### Development (docker-compose)

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

Then:

```python
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

```python
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.

```python
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:

| Class                           | Behavior                                                  |
| ------------------------------- | --------------------------------------------------------- |
| `JsonMetadataSerializer`        | Strings pass through; others → JSON with `"json:"` prefix |
| `PassthroughMetadataSerializer` | All values → `str()` (no deserialization)                 |

---

## Builder API Reference

### `StorageBuilder(client)`

| Method                         | Returns              | Description                |
| ------------------------------ | -------------------- | -------------------------- |
| `.bucket(name)`                | `BucketOperation`    | Bucket-scoped operations   |
| `.object(bucket, key)`         | `ObjectOperation`    | Single-object operations   |
| `.multipart(bucket, key)`      | `MultipartOperation` | Multipart upload lifecycle |
| `.connect()` / `.disconnect()` | —                    | Delegate to client         |
| `.ping()`                      | `bool`               | Health check               |

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

### `BucketOperation`

| Method                                   | Returns         | Description                   |
| ---------------------------------------- | --------------- | ----------------------------- |
| `.create()`                              | —               | Create bucket (idempotent)    |
| `.delete()`                              | —               | Delete bucket (must be empty) |
| `.exists()`                              | `bool`          | Check existence               |
| `.list_objects(prefix, max_keys, token)` | `ObjectListing` | Paginated listing             |

### `ObjectOperation`

| Method                                             | Returns         | Description                   |
| -------------------------------------------------- | --------------- | ----------------------------- |
| `.put(data, metadata, content_type)`               | `StorageObject` | Shorthand for `put_from_data` |
| `.put_from_data(data, metadata, content_type)`     | `StorageObject` | Upload raw bytes              |
| `.put_from_path(path, metadata, content_type)`     | `StorageObject` | Read local file and upload    |
| `.put_from_source(source, metadata, content_type)` | `StorageObject` | Upload from async stream      |
| `.get()`                                           | `StorageObject` | Retrieve with data            |
| `.head()`                                          | `StorageObject` | Retrieve metadata only        |
| `.delete()`                                        | —               | Delete object                 |
| `.copy_to(dest_bucket, dest_key)`                  | `StorageObject` | Copy to another location      |
| `.presigned_get(expires_in)`                       | `str`           | Time-limited download URL     |
| `.presigned_put(expires_in)`                       | `str`           | Time-limited upload URL       |

### `MultipartOperation`

| Method                                       | Returns           | Description     |
| -------------------------------------------- | ----------------- | --------------- |
| `.create(metadata, content_type)`            | `str` (upload_id) | Initiate upload |
| `.upload_part(upload_id, part_number, data)` | `dict`            | Upload one part |
| `.complete(upload_id, parts)`                | `StorageObject`   | Finalize upload |
| `.abort(upload_id)`                          | —                 | Cancel upload   |

---

## 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**.

```bash
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.
