s-storagekit

Объектное хранилище для Skillery: портабельное async-API для key→bytes операций с реализациями S3-compatible протокола (MinIO) и вспомогательными утилитами нормализации изображений (Pillow).

Установка

pip install s-storagekit>=0.1.0

Быстрый старт

Использование MinIO

from storagekit import MinioObjectStorage, ObjectNotFoundError

# Инициализация
storage = MinioObjectStorage(
    endpoint="minio:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    bucket="my-bucket",
    secure=False,  # http в dev, https в prod
)

# Запись объекта
await storage.put("avatars/user_1/photo.jpg", image_bytes, "image/jpeg")

# Чтение объекта
data, content_type = await storage.get("avatars/user_1/photo.jpg")

# Удаление объекта
await storage.delete("avatars/user_1/photo.jpg")

Обработка изображений

from storagekit import (
    normalize_image,
    normalize_square_avatar,
    square_avatar_variants,
)

# Квадратный аватар одного размера
jpeg_bytes, ct = normalize_square_avatar(raw_image, size=256)
await storage.put(f"avatars/user_1/{token}.jpg", jpeg_bytes, ct)

# Несколько вариантов размеров (responsive)
variants = square_avatar_variants(raw_image, sizes=[64, 128, 256])
for size, blob in variants:
    key = f"avatars/user_1/{token}_{size}.jpg"
    await storage.put(key, blob, "image/jpeg")

# Масштабирование обложки (сохраняет пропорции)
cover_data, ct = normalize_image(raw_image, max_px=1200)
await storage.put(f"skills/42/cover_{token}.jpg", cover_data, ct)

Тестирование

import pytest
from storagekit.testing import InMemoryObjectStorage, fake_image_bytes

@pytest.fixture
def storage():
    return InMemoryObjectStorage()

@pytest.mark.asyncio
async def test_avatar_upload(storage):
    raw = fake_image_bytes(100, 100)
    jpeg_bytes, _ = normalize_square_avatar(raw, size=256)
    
    await storage.put("avatars/test.jpg", jpeg_bytes, "image/jpeg")
    data, ct = await storage.get("avatars/test.jpg")
    
    assert ct == "image/jpeg"
    assert len(data) > 0

Presigned URL (без auth)

from storagekit import PresignedUrlGenerator

gen = PresignedUrlGenerator(
    endpoint="minio:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    bucket="my-bucket",
)

# Генерировать URL с TTL (по умолчанию 1 час)
url = await gen.generate_presigned_get_url("avatars/user_1/photo.jpg")
# Клиент может скачать файл по этому URL без токена

API

Протоколы

class ObjectStorage(Protocol):
    """Async key→bytes хранилище."""
    
    async def put(self, key: str, data: bytes, content_type: str) -> None:
        """Записать/перезаписать объект."""
        ...
    
    async def get(self, key: str) -> tuple[bytes, str]:
        """Вернуть (data, content_type). Бросает ObjectNotFoundError."""
        ...
    
    async def delete(self, key: str) -> None:
        """Удалить объект. Отсутствие не ошибка (idempotent)."""
        ...

Исключения

class ObjectNotFoundError(Exception):
    """Объект не найден. Имеет атрибут .key."""
    def __init__(self, key: str) -> None: ...

Реализации

# S3-compatible хранилище (prod/dev)
class MinioObjectStorage:
    def __init__(
        self,
        *,
        endpoint: str,        # host:port БЕЗ схемы
        access_key: str,
        secret_key: str,
        bucket: str,
        secure: bool = False, # http (dev) vs https (prod)
    ) -> None: ...

Image-утилиты

# Список допустимых MIME-типов входных файлов
ALLOWED_IMAGE_CONTENT_TYPES: frozenset[str]

# Квадратный аватар (center cover-crop) → JPEG
def normalize_square_avatar(data: bytes, *, size: int) -> tuple[bytes, str]: ...

# Несколько вариантов размеров одного аватара
def square_avatar_variants(
    data: bytes, *, sizes: list[int]
) -> list[tuple[int, bytes]]: ...

# Масштабирование по длинной стороне (сохраняет пропорции)
def normalize_image(data: bytes, *, max_px: int) -> tuple[bytes, str]: ...

Утилиты

# Канонический ключ snapshot'а версии
def snapshot_key(*, skill_id: int | str, semver: str) -> str:
    """Возвращает f'skills/{skill_id}/versions/{semver}.tar.gz'."""

Тестирование

# Фейк-хранилище в памяти
class InMemoryObjectStorage:
    async def put(self, key: str, data: bytes, content_type: str) -> None: ...
    async def get(self, key: str) -> tuple[bytes, str]: ...
    async def delete(self, key: str) -> None: ...

# Генерировать тестовый PNG-файл
def fake_image_bytes(width: int = 100, height: int = 100) -> bytes: ...

Параметризация

Зависимости

• minio>=7.0 — S3 SDK
• anyio>=3.0 — async threading
• Pillow>=10.0 — image processing

Лицензия

MIT