Source code for signal_dataset.storage.backends.s3.options

"""Tunable settings for the S3 backend.

These live here rather than in :mod:`signal_dataset.config` because the storage
domain may not import it (see ``tests/test_architecture.py``), and because
endpoints and retry policy are transport concerns rather than dataset ones.

Credentials are deliberately absent. Resolution is boto3's own chain —
environment, shared config, instance or container role, web identity — matching
how the GCS backend defers to Application Default Credentials. This package
neither reads nor stores credentials.

To use these with ``sds.open`` and friends, either pass a configured store as
``object_store=``, or register a configured backend once::

    from signal_dataset.storage import default_registry
    from signal_dataset.storage.backends.s3 import S3Options, s3_backend

    default_registry().register(
        s3_backend(S3Options(endpoint_url="https://minio.example", addressing_style="path")),
        replace=True,
    )
"""

from __future__ import annotations

from dataclasses import dataclass

# A single PutObject is capped by S3 at 5 GiB. Anything larger needs multipart,
# which cannot express create-only preconditions the same way.
MAX_SINGLE_PART_BYTES = 5 * 1024**3

DEFAULT_MAX_ATTEMPTS = 5
DEFAULT_RETRY_MODE = "standard"
DEFAULT_CONFLICT_ATTEMPT_LIMIT = 5
DEFAULT_CONFLICT_RETRY_BASE_SECONDS = 0.05
DEFAULT_CONFLICT_RETRY_MAX_SECONDS = 1.0
DEFAULT_ETAG_MEMO_SIZE = 128

RETRY_MODES = ("legacy", "standard", "adaptive")
ADDRESSING_STYLES = ("virtual", "path")




@dataclass(frozen=True, slots=True)
class S3Options:
    """Transport settings for :class:`S3ObjectStore`."""

    region_name: str | None = None
    endpoint_url: str | None = None
    profile_name: str | None = None
    addressing_style: str | None = None
    max_attempts: int = DEFAULT_MAX_ATTEMPTS
    retry_mode: str = DEFAULT_RETRY_MODE
    conflict_attempt_limit: int = DEFAULT_CONFLICT_ATTEMPT_LIMIT
    """Total conditional-write attempts, not retries: 1 means no retry."""
    conflict_retry_base_seconds: float = DEFAULT_CONFLICT_RETRY_BASE_SECONDS
    conflict_retry_max_seconds: float = DEFAULT_CONFLICT_RETRY_MAX_SECONDS
    etag_memo_size: int = DEFAULT_ETAG_MEMO_SIZE
    max_single_part_bytes: int = MAX_SINGLE_PART_BYTES

    def __post_init__(self) -> None:
        for name in (
            "max_attempts",
            "conflict_attempt_limit",
            "etag_memo_size",
            "max_single_part_bytes",
        ):
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int) or value < 1:
                raise ValueError(f"{name} must be a positive integer")
        if self.max_single_part_bytes > MAX_SINGLE_PART_BYTES:
            raise ValueError(
                f"max_single_part_bytes must not exceed the S3 limit of {MAX_SINGLE_PART_BYTES}"
            )
        for name in ("conflict_retry_base_seconds", "conflict_retry_max_seconds"):
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int | float) or value < 0:
                raise ValueError(f"{name} must be a nonnegative number")
        if self.conflict_retry_max_seconds < self.conflict_retry_base_seconds:
            raise ValueError("conflict_retry_max_seconds must not be below the base delay")
        if self.retry_mode not in RETRY_MODES:
            raise ValueError(f"retry_mode must be one of {RETRY_MODES}")
        if self.addressing_style is not None and self.addressing_style not in ADDRESSING_STYLES:
            raise ValueError(f"addressing_style must be one of {ADDRESSING_STYLES} or None")
        for name in ("region_name", "endpoint_url", "profile_name"):
            value = getattr(self, name)
            if value is not None and not value:
                raise ValueError(f"{name} must be non-empty or None")