rasteret.cloud

Cloud provider configuration and URL signing.

cloud

Classes

CloudConfig dataclass

CloudConfig(
    provider: str,
    requester_pays: bool = False,
    region: str = "us-west-2",
    url_patterns: dict[str, str] = dict(),
)

Storage configuration for a data source.

Built-in configs may be pre-registered for some well-known data sources that need URL rewriting or requester-pays access. Register your own via :meth:register::

CloudConfig.register("my-collection", CloudConfig(
    provider="aws", requester_pays=True, region="eu-west-1",
))

Functions

register classmethod

register(collection_id: str, config: CloudConfig) -> None

Register a cloud config for a collection id.

Source code in src/rasteret/cloud.py

@classmethod
def register(cls, collection_id: str, config: CloudConfig) -> None:
    """Register a cloud config for a collection id."""
    cls._configs[collection_id.lower()] = config

get_config classmethod

get_config(data_source: str) -> CloudConfig | None

Look up cloud config for data_source.

Source code in src/rasteret/cloud.py

@classmethod
def get_config(cls, data_source: str) -> CloudConfig | None:
    """Look up cloud config for *data_source*."""
    return cls._configs.get(data_source.lower())

StorageBackend

Bases: Protocol

Minimal protocol for range-based reads from cloud storage.

Implement this to plug in a custom I/O backend (e.g. obstore, fsspec, or a mocked reader for tests).

ObstoreBackend

ObstoreBackend(
    store: object,
    url_prefix: str = "",
    client_options: dict[str, object] | None = None,
)

StorageBackend backed by the obstore library.

Wraps obstore.get_range_async / obstore.get_ranges_async. Pass any obstore.store.*Store instance.

Parameters:

Name	Type	Description	Default
store	obstore Store	Any obstore store (S3Store, HTTPStore, etc.).	required
url_prefix	str	URL prefix to strip before passing paths to obstore. obstore expects paths relative to the store root, but rasteret works with full URLs. For example, if COG URLs look like https://sentinel-cogs.s3.us-west-2.amazonaws.com/sentinel-s2-l2a-cogs/... and the store is HTTPStore.from_url("https://sentinel-cogs.s3.us-west-2.amazonaws.com/"), set url_prefix="https://sentinel-cogs.s3.us-west-2.amazonaws.com/".	''
client_options	dict	ClientConfig options forwarded to HTTPStore.from_url when constructing the store. Ignored if a pre-built store is provided.	None

Source code in src/rasteret/cloud.py

def __init__(
    self,
    store: object,
    url_prefix: str = "",
    client_options: dict[str, object] | None = None,
) -> None:
    self._store = store
    self._url_prefix = url_prefix
    self._client_options = client_options

Functions

rewrite_url

rewrite_url(url: str, config: CloudConfig | None) -> str

Apply URL pattern rewrites without presigning.

Pure string transformation. Returns the original URL if no patterns match or config is None.

Source code in src/rasteret/cloud.py

def rewrite_url(url: str, config: CloudConfig | None) -> str:
    """Apply URL pattern rewrites without presigning.

    Pure string transformation.  Returns the original URL if no
    patterns match or *config* is ``None``.
    """
    if config and config.url_patterns and isinstance(url, str):
        for http_pattern, s3_pattern in config.url_patterns.items():
            if url.startswith(http_pattern):
                return url.replace(http_pattern, s3_pattern)
    return url

s3_overrides_from_config

s3_overrides_from_config(
    config: CloudConfig | None,
) -> dict[str, dict[str, str]]

Derive per-bucket S3Store config from a :class:CloudConfig.

Returns a mapping of {bucket: {config_key: value}} suitable for passing to :class:~rasteret.fetch.cog._AutoObstoreBackend.

Example: CloudConfig(requester_pays=True, url_patterns={...}) yields {"usgs-landsat": {"request_payer": "true", "region": "us-west-2"}}.

Source code in src/rasteret/cloud.py

def s3_overrides_from_config(config: CloudConfig | None) -> dict[str, dict[str, str]]:
    """Derive per-bucket S3Store config from a :class:`CloudConfig`.

    Returns a mapping of ``{bucket: {config_key: value}}`` suitable for
    passing to :class:`~rasteret.fetch.cog._AutoObstoreBackend`.

    Example: ``CloudConfig(requester_pays=True, url_patterns={...})``
    yields ``{"usgs-landsat": {"request_payer": "true", "region": "us-west-2"}}``.
    """
    if config is None:
        return {}
    overrides: dict[str, dict[str, str]] = {}
    for s3_pattern in (config.url_patterns or {}).values():
        # Extract bucket from s3://bucket/ pattern
        if s3_pattern.startswith("s3://"):
            bucket = s3_pattern.split("/")[2]
            bucket_config: dict[str, str] = {"region": config.region}
            if config.requester_pays:
                bucket_config["request_payer"] = "true"
            else:
                bucket_config["skip_signature"] = "true"
            overrides[bucket] = bucket_config
    return overrides

backend_config_from_cloud_config

backend_config_from_cloud_config(
    config: CloudConfig | None,
) -> dict

Extract :func:~rasteret.fetch.cog._create_obstore_backend kwargs from a CloudConfig.

Source code in src/rasteret/cloud.py

def backend_config_from_cloud_config(config: CloudConfig | None) -> dict:
    """Extract :func:`~rasteret.fetch.cog._create_obstore_backend` kwargs from a CloudConfig."""
    if config is None:
        return {}
    result: dict = {}
    overrides = s3_overrides_from_config(config)
    if overrides:
        result["s3_overrides"] = overrides
    if config.url_patterns:
        result["url_patterns"] = config.url_patterns
    # When requester_pays is True, set default_s3_config so that buckets
    # discovered at runtime (e.g. from s3:// sibling assets in STAC)
    # are configured correctly even if they don't appear in url_patterns.
    if config.requester_pays:
        result["default_s3_config"] = {
            "request_payer": "true",
            "region": config.region,
        }
    return result