s3

storage cloud · maintainer machinable-org · upstream

Records in an S3 or S3-compatible bucket, read on demand.

Install

Declare the remote in your project's provider; the module is fetched on first use:

# interface/project.py — your project's provider
from machinable import Project


class MyProject(Project):
    def on_resolve_remotes(self):
        return {
            "s3": "url+https://raw.githubusercontent.com/machinable-org/machinable/f672c4f238c207f34da2e3919014e436970d08d8/integrations/s3/s3.py",
        }

Runtime dependencies: pip install obstore

The module is fetched once and cached in interface/remotes/; machinable fetch s3 downloads it without importing so you can inspect the code first. See Resolving remotes for how pulling, pinning, and updating work.

Requires obstore and an S3 bucket (AWS S3, or an S3-compatible service such as MinIO or Cloudflare R2).

Results in a bucket

Connect the storage and the bucket is the home of the project's records: backed up, reachable from any machine, with the local folder only a working copy that can be lost.

from machinable import get

with get("s3", {"url": "s3://my-bucket/lab/machinable", "region": "us-east-1"}):
    get("my_analysis", {"session": 12}).launch()

Runs write a local working copy under directory (default ./storage, in the same layout as the bucket under the URL's prefix). Dispatch puts the run-record in the bucket when it starts and each status marker as it is written (heartbeats at most every half liveness window), pushes the run's files when it ends, finished_at after them. Outside a run, writes through machinable (save_file, ...) go to the bucket as they are made. Delete the working copy and the next read fetches the record again.

Pull the module into a project through on_resolve_remotes, pinned to a commit.

Config	Meaning
url	s3://bucket/prefix
region	the bucket's region
endpoint	an S3-compatible endpoint (http://localhost:9000 for MinIO)
directory	local working copies (default ./storage)
catalog_max_bytes	files up to this size are a record's catalog (default 1 MiB)
credential_provider	module:function returning credentials, see below
conditional_writes	false for a store without If-None-Match/If-Match (default true)

Credentials

In order of preference:

1. obstore's default chain: the environment (AWS_ACCESS_KEY_ID, ...), a profile, or the instance role through IMDS on a cloud machine. Nothing to configure.
2. A credential provider for short-lived credentials that refresh (STS credentials from a sign-in): a callable returning {"access_key_id", "secret_access_key", "token", "expires_at"}, which obstore calls again before expires_at. Pass it to the constructor, S3(config, credential_provider=fn), or name it in the config as "package.module:function" (a process that connects the storage from its config, such as machinable serve, imports it from there).
3. Explicit access_key_id, secret_access_key and session_token in the config (avoid in code you commit).

IAM

One user's project needs s3:GetObject, s3:PutObject, s3:DeleteObject on the prefix and s3:ListBucket on the bucket, limited to the prefix:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
      "Resource": "arn:aws:s3:::my-bucket/lab/machinable/*"
    },
    {
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::my-bucket",
      "Condition": {"StringLike": {"s3:prefix": ["lab/machinable", "lab/machinable/*"]}}
    }
  ]
}

Conditional writes need no extra permission. Large files are uploaded in parts; a lifecycle rule that aborts incomplete multipart uploads after a day keeps a crashed upload from costing storage.

Reading on demand

A machine that did not make a record sees it through the bucket's catalog, without downloading its data:

with get("s3", {"url": "s3://my-bucket/lab/machinable"}) as storage:
    ingested, position = storage.sync()      # what was added since last time
    run = get("my_analysis", {"session": 12})
    run.load_file("metrics.json")            # small files are already here
    run.local_directory("model.pt")          # a large one is downloaded now
    run.local_directory("analyzer")          # a folder: the files it lacks
    run.fetch()                              # everything

sync(since) reads the bucket's log of new and changed records (.catalog/<time>-<uuid> under the prefix, written when a record is first pushed and again when it changes) after since, indexes each new record's catalog (its files up to catalog_max_bytes, its markers and its edge log) and refreshes each changed one. Such records are bytes_missing until the rest is read. storage.follow(15) keeps a machine's index current by syncing every 15 s in the background (following a home). Index.get().reindex(storage) rebuilds an index from the whole bucket the same way.

storage.delete(uuid) removes a record (and its runs) from the bucket; the working copy stays.

Several machines on one bucket

Machines that connect the same bucket share its catalog, and an interface dispatched on several of them at once runs once, as for processes sharing a folder (concurrent dispatch):

• a finished run in the bucket is a cache hit everywhere (cached() reason "finished elsewhere"), its results read on demand;
• two machines dispatching at once: one claims the interface by creating <identity>/.claims/<predicate> beside it with If-None-Match: *; the other waits on that run (or, with Execution().wait_elsewhere(False), reports it in execution.elsewhere) and ends cached on it;
• a claimant that stops beating for the liveness window (90 s by default, MACHINABLE_HEARTBEAT_WINDOW) is taken over by exactly one waiter, which replaces the claim with If-Match on its ETag and marks the dead run-record abandoned_at.

Every machine gives an interface the same record id. Each machine keeps its own index; nothing but the bucket is shared.

This needs a store that honours conditional writes on PUT: AWS S3 (both headers since late 2024), MinIO (recent releases) and Cloudflare R2 do. For a store that does not (check yours: a second PUT with If-None-Match: * must fail with 412), set conditional_writes: false; dispatch then still finds runs finished elsewhere, but two machines dispatching at once can both run.

Configuration

Option	Type	Default
url	str	Field('???')
directory	str	'./storage'
region	str | None	None
endpoint	str | None	None
access_key_id	str | None	None
secret_access_key	str | None	None
session_token	str | None	None
credential_provider	str | None	None
catalog_max_bytes	int	1024 * 1024
conditional_writes	bool	True

Source

integrations/s3/s3.py

s3.py

"""S3: records in an S3 or S3-compatible bucket (via obstore)."""

from collections.abc import Callable
from typing import Any
from urllib.parse import urlparse

from pydantic import BaseModel, Field

from machinable import Storage

# objects up to this size are read whole, larger ones in chunks
_WHOLE = 8 * 1024 * 1024


class S3(Storage):
    """A home in a bucket: ``s3://bucket/prefix``."""

    catalog_log = True

    class Config(BaseModel):
        """Where the records live, and how to reach the bucket.

        Credentials come from obstore's default chain (environment, profile,
        instance role) unless given here or by a ``credential_provider``
        (``module:function``, or a callable passed to the constructor)
        returning refreshing credentials. ``endpoint`` points at an
        S3-compatible service; ``conditional_writes: false`` for one that
        does not honour ``If-None-Match``/``If-Match`` (no exclusive claims).
        """

        url: str = Field("???")
        directory: str = "./storage"
        region: str | None = None
        endpoint: str | None = None
        access_key_id: str | None = None
        secret_access_key: str | None = None
        session_token: str | None = None
        credential_provider: str | None = None
        catalog_max_bytes: int = 1024 * 1024
        conditional_writes: bool = True

    def __init__(
        self, version: Any = None, credential_provider: Callable | None = None
    ):
        super().__init__(version=version)
        self._credential_provider = credential_provider
        self._store: Any = None

    @property
    def id(self) -> str:
        """Identifier for this storage provider."""
        return "s3"

    @property
    def store(self) -> Any:
        """The obstore ``S3Store`` of ``url`` (made on first use)."""
        if self._store is None:
            from obstore.store import S3Store

            parsed = urlparse(self.config.url)
            if parsed.scheme != "s3" or not parsed.netloc:
                raise ValueError(
                    f"Expected s3://bucket/prefix, got '{self.config.url}'"
                )
            keys = ("region", "endpoint", "access_key_id", "secret_access_key")
            config = {k: getattr(self.config, k) for k in (*keys, "session_token")}
            provider = self._credential_provider
            if provider is None and self.config.credential_provider:
                from machinable.config import import_object_by_path

                provider = import_object_by_path(self.config.credential_provider)
            http = str(self.config.endpoint or "").startswith("http://")
            self._store = S3Store(
                parsed.netloc,
                prefix=parsed.path.strip("/") or None,
                config={k: v for k, v in config.items() if v is not None},
                client_options={"allow_http": True} if http else None,
                credential_provider=provider,
            )
        return self._store

    def root_uri(self) -> str:
        """The bucket URL records are laid out under."""
        return self.config.url.rstrip("/")

    def owns(self, uri: str | None) -> bool:
        """Whether ``uri`` is under this storage's URL."""
        root = self.root_uri()
        return uri is not None and (uri == root or uri.startswith(root + "/"))

    def _key(self, location: str, name: str = "") -> str:
        relative = location[len(self.root_uri()) :].strip("/")
        return "/".join(p for p in (relative, name) if p)

    def list(self, location: str, recursive: bool = True):
        """The objects under ``location`` (a paginated LIST)."""
        import obstore

        prefix = self._key(location)
        base = prefix + "/" if prefix else ""
        if recursive:
            objects = [
                m for batch in obstore.list(self.store, prefix or None) for m in batch
            ]
            directories = []
        else:
            found = obstore.list_with_delimiter(self.store, prefix or None)
            objects = found["objects"]
            directories = [
                (p[len(base) :] + "/", 0, None) for p in found["common_prefixes"]
            ]
        files = [(m["path"][len(base) :], m["size"], m["e_tag"]) for m in objects]
        return sorted(files + directories)

    def read(self, location: str, name: str):
        """GET an object: bytes, or chunks for a large one; None if absent."""
        import obstore

        try:
            result = obstore.get(self.store, self._key(location, name))
        except FileNotFoundError:
            return None
        if result.meta["size"] <= _WHOLE:
            return bytes(result.bytes())
        return (bytes(chunk) for chunk in result)

    def write(self, location: str, name: str, data) -> None:
        """PUT an object (multipart for a large stream)."""
        import obstore

        obstore.put(self.store, self._key(location, name), data)

    def remove(self, location: str, name: str) -> None:
        """DELETE an object."""
        import obstore

        try:
            obstore.delete(self.store, self._key(location, name))
        except FileNotFoundError:
            pass

    def exclusive_claims(self) -> bool:
        """Whether the bucket honours conditional writes."""
        return self.config.conditional_writes

    def read_versioned(self, location: str) -> tuple[bytes, str | None] | None:
        """GET an object with its ETag."""
        import obstore

        try:
            result = obstore.get(self.store, self._key(location))
        except FileNotFoundError:
            return None
        return bytes(result.bytes()), result.meta.get("e_tag")

    def create_if_absent(self, location: str, data: bytes) -> bool:
        """PUT with ``If-None-Match: *``."""
        import obstore
        from obstore.exceptions import AlreadyExistsError, PreconditionError

        try:
            obstore.put(self.store, self._key(location), data, mode="create")
        except (AlreadyExistsError, PreconditionError):
            return False
        return True

    def replace_if_unchanged(self, location: str, data: bytes, version: str) -> bool:
        """PUT with ``If-Match`` on the ETag ``version``."""
        import obstore
        from obstore.exceptions import PreconditionError

        try:
            obstore.put(self.store, self._key(location), data, mode={"e_tag": version})
        except PreconditionError:
            return False
        return True